tests/unit/test_watcher.py

   1 # Copyright (C) 2021 Wind River Systems, Inc.\r
   2 #\r
   3 #  Licensed under the Apache License, Version 2.0 (the "License");\r
   4 #  you may not use this file except in compliance with the License.\r
   5 #  You may obtain a copy of the License at\r
   6 #\r
   7 #      http://www.apache.org/licenses/LICENSE-2.0\r
   8 #\r
   9 #  Unless required by applicable law or agreed to in writing, software\r
  10 #  distributed under the License is distributed on an "AS IS" BASIS,\r
  11 #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
  12 #  See the License for the specific language governing permissions and\r
  13 #  limitations under the License.\r
  14 \r
  15 import time\r
  16 from datetime import datetime\r
  17 import json\r
  18 from typing import List\r
  19 from o2ims.domain.resource_type import ResourceTypeEnum\r
  20 from o2ims.service.client.base_client import BaseClient\r
  21 from o2ims.domain import ocloud\r
  22 from o2ims import config\r
  23 import uuid\r
  24 from o2ims.service.watcher.base import BaseWatcher, OcloudWatcher\r
  25 from o2ims.domain import stx_object as ocloudModel\r
  26 from o2ims.adapter.ocloud_repository import OcloudRepository\r
  27 from o2ims.domain.stx_repo import StxObjectRepository\r
  28 from o2ims.service.watcher import worker\r
  29 from o2ims.service.unit_of_work import AbstractUnitOfWork\r
  30 \r
  31 \r
  32 class FakeOcloudClient(BaseClient):\r
  33     def __init__(self):\r
  34         super().__init__()\r
  35         fakeCloud = ocloudModel.StxGenericModel(ResourceTypeEnum.OCLOUD)\r
  36         fakeCloud.id = uuid.uuid4()\r
  37         fakeCloud.name = 'stx1'\r
  38         fakeCloud.content = json.dumps({})\r
  39         fakeCloud.createtime = datetime.now()\r
  40         fakeCloud.updatetime = datetime.now()\r
  41         fakeCloud.hash = str(hash((fakeCloud.id, fakeCloud.updatetime)))\r
  42         self.fakeCloud = fakeCloud\r
  43 \r
  44     def _get(self, id) -> ocloudModel.StxGenericModel:\r
  45         return self.fakeCloud\r
  46 \r
  47     def _list(self):\r
  48         return [self.fakeCloud]\r
  49 \r
  50 \r
  51 class FakeOcloudRepo(OcloudRepository):\r
  52     def __init__(self):\r
  53         super().__init__()\r
  54         self.oclouds = []\r
  55 \r
  56     def _add(self, ocloud: ocloud.Ocloud):\r
  57         self.oclouds.append(ocloud)\r
  58 \r
  59     def _get(self, ocloudid) -> ocloud.Ocloud:\r
  60         filtered = [o for o in self.oclouds if o.id == ocloudid]\r
  61         return filtered.pop()\r
  62 \r
  63     def _list(self) -> List[ocloud.Ocloud]:\r
  64         return [x for x in self.oclouds]\r
  65 \r
  66     def _update(self, ocloud: ocloud.Ocloud):\r
  67         filtered = [o for o in self.oclouds if o.id == ocloud.id]\r
  68         assert len(filtered) == 1\r
  69         ocloud1 = filtered.pop()\r
  70         ocloud1.update_by(ocloud)\r
  71 \r
  72 \r
  73 \r
  74 class FakeStxObjRepo(StxObjectRepository):\r
  75     def __init__(self):\r
  76         super().__init__()\r
  77         self.oclouds = []\r
  78 \r
  79     def _add(self, ocloud: ocloud.Ocloud):\r
  80         self.oclouds.append(ocloud)\r
  81 \r
  82     def _get(self, ocloudid) -> ocloud.Ocloud:\r
  83         filtered = [o for o in self.oclouds if o.id == ocloudid]\r
  84         return filtered.pop()\r
  85 \r
  86     def _list(self, type: ResourceTypeEnum) -> List[ocloud.Ocloud]:\r
  87         return [x for x in self.oclouds]\r
  88 \r
  89     def _update(self, ocloud: ocloud.Ocloud):\r
  90         filtered = [o for o in self.oclouds if o.id == ocloud.id]\r
  91         assert len(filtered) == 1\r
  92         ocloud1 = filtered.pop()\r
  93         ocloud1.update_by(ocloud)\r
  94 \r
  95 \r
  96 class FakeUnitOfWork(AbstractUnitOfWork):\r
  97     def __init__(self):\r
  98         pass\r
  99 \r
 100     def __enter__(self):\r
 101         # self.session = self.session_factory()  # type: Session\r
 102         # self.oclouds = OcloudSqlAlchemyRepository(self.session)\r
 103         self.stxobjects = FakeStxObjRepo()\r
 104         return super().__enter__()\r
 105 \r
 106     def __exit__(self, *args):\r
 107         super().__exit__(*args)\r
 108         # self.session.close()\r
 109 \r
 110     def _commit(self):\r
 111         pass\r
 112         # self.session.commit()\r
 113 \r
 114     def rollback(self):\r
 115         pass\r
 116         # self.session.rollback()\r
 117 \r
 118 \r
 119 def test_probe_new_ocloud():\r
 120     # fakeRepo = FakeOcloudRepo()\r
 121     fakeuow = FakeUnitOfWork()\r
 122     fakeClient = FakeOcloudClient()\r
 123     ocloudwatcher = OcloudWatcher(fakeClient, fakeuow)\r
 124     ocloudwatcher.probe()\r
 125     assert len(fakeuow.stxobjects.oclouds) == 1\r
 126     assert fakeuow.stxobjects.oclouds[0].name == "stx1"\r
 127 \r
 128 \r
 129 def test_watchers_worker():\r
 130     testedworker = worker.PollWorker()\r
 131 \r
 132     class FakeOCloudWatcher(BaseWatcher):\r
 133         def __init__(self, client: BaseClient,\r
 134                      repo: OcloudRepository) -> None:\r
 135             super().__init__(client)\r
 136             self.fakeOcloudWatcherCounter = 0\r
 137             self._client = client\r
 138             self._repo = repo\r
 139 \r
 140         def _targetname(self):\r
 141             return "fakeocloudwatcher"\r
 142 \r
 143         def _probe(self):\r
 144             self.fakeOcloudWatcherCounter += 1\r
 145             # hacking to stop the blocking sched task\r
 146             if self.fakeOcloudWatcherCounter > 2:\r
 147                 testedworker.stop()\r
 148 \r
 149 \r
 150     # fakeRepo = FakeOcloudRepo()\r
 151     fakeuow = FakeUnitOfWork()\r
 152 \r
 153     fakeClient = FakeOcloudClient()\r
 154     fakewatcher = FakeOCloudWatcher(fakeClient, fakeuow)\r
 155 \r
 156     testedworker.set_interval(1)\r
 157     testedworker.add_watcher(fakewatcher)\r
 158     assert fakewatcher.fakeOcloudWatcherCounter == 0\r
 159 \r
 160     count1 = fakewatcher.fakeOcloudWatcherCounter\r
 161     testedworker.start()\r
 162     time.sleep(20)\r
 163     assert fakewatcher.fakeOcloudWatcherCounter > count1\r
 164 \r
 165     # assumed hacking: probe has stopped the sched task\r
 166     count3 = fakewatcher.fakeOcloudWatcherCounter\r
 167     time.sleep(3)\r
 168     assert fakewatcher.fakeOcloudWatcherCounter == count3\r