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 from datetime import datetime\r
  16 import json\r
  17 from typing import List\r
  18 from o2ims.service.client.base_client import BaseClient\r
  19 import pytest\r
  20 from o2ims.domain import ocloud\r
  21 from o2ims import config\r
  22 import uuid\r
  23 from o2ims.service.watcher.base import OcloudWather\r
  24 from o2ims.domain import stx_object as ocloudModel\r
  25 from o2ims.adapter.ocloud_repository import OcloudRepository\r
  26 \r
  27 class FakeOcloudClient(BaseClient):\r
  28     def __init__(self):\r
  29         super().__init__()\r
  30         fakeCloud = ocloudModel.StxGenericModel()\r
  31         fakeCloud.id = uuid.uuid4()\r
  32         fakeCloud.name = 'stx1'\r
  33         fakeCloud.content = json.dumps({})\r
  34         fakeCloud.createtime = datetime.now()\r
  35         fakeCloud.updatetime = datetime.now\r
  36         self.fakeCloud = fakeCloud\r
  37 \r
  38     def _get(self, id) -> ocloudModel.StxGenericModel:\r
  39         return self.fakeCloud\r
  40 \r
  41     def _list(self):\r
  42         return [self.fakeCloud]\r
  43 \r
  44 class FakeOcloudRepo(OcloudRepository):\r
  45     def __init__(self):\r
  46         super().__init__()\r
  47         self.oclouds = []\r
  48 \r
  49     def _add(self, ocloud: ocloud.Ocloud):\r
  50         self.oclouds.append(ocloud)\r
  51 \r
  52     def _get(self, ocloudid) -> ocloud.Ocloud:\r
  53         filtered = [o for o in self.oclouds if o.id == ocloudid]\r
  54         return filtered.pop()\r
  55 \r
  56     def _list(self) -> List[ocloud.Ocloud]:\r
  57         return [x for x in self.oclouds]\r
  58 \r
  59     def _update(self, ocloud: ocloud.Ocloud):\r
  60         filtered = [o for o in self.oclouds if o.id == ocloud.id]\r
  61         assert len(filtered) == 1\r
  62         ocloud1 = filtered.pop()\r
  63         ocloud1.update_by(ocloud)\r
  64 \r
  65 def test_probe_new_ocloud():\r
  66     fakeRepo = FakeOcloudRepo()\r
  67     fakeClient = FakeOcloudClient()\r
  68     ocloudwatcher = OcloudWather(fakeClient, fakeRepo)\r
  69     ocloudwatcher.probe()\r
  70     assert len(fakeRepo.oclouds) == 1\r
  71     assert fakeRepo.oclouds[0].name == "stx1"\r