+# Copyright (C) 2021 Wind River Systems, Inc.\r
+#\r
+# Licensed under the Apache License, Version 2.0 (the "License");\r
+# you may not use this file except in compliance with the License.\r
+# You may obtain a copy of the License at\r
+#\r
+# http://www.apache.org/licenses/LICENSE-2.0\r
+#\r
+# Unless required by applicable law or agreed to in writing, software\r
+# distributed under the License is distributed on an "AS IS" BASIS,\r
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
+# See the License for the specific language governing permissions and\r
+# limitations under the License.\r
+\r
+from datetime import datetime\r
+import json\r
+from typing import List\r
+from o2ims.service.client.base_client import BaseClient\r
+import pytest\r
+from o2ims.domain import ocloud\r
+from o2ims import config\r
+import uuid\r
+from o2ims.service.watcher.base import OcloudWather\r
+from o2ims.domain import stx_object as ocloudModel\r
+from o2ims.adapter.ocloud_repository import OcloudRepository\r
+\r
+class FakeOcloudClient(BaseClient):\r
+ def __init__(self):\r
+ super().__init__()\r
+ fakeCloud = ocloudModel.StxGenericModel()\r
+ fakeCloud.id = uuid.uuid4()\r
+ fakeCloud.name = 'stx1'\r
+ fakeCloud.content = json.dumps({})\r
+ fakeCloud.createtime = datetime.now()\r
+ fakeCloud.updatetime = datetime.now\r
+ self.fakeCloud = fakeCloud\r
+\r
+ def _get(self, id) -> ocloudModel.StxGenericModel:\r
+ return self.fakeCloud\r
+\r
+ def _list(self):\r
+ return [self.fakeCloud]\r
+\r
+class FakeOcloudRepo(OcloudRepository):\r
+ def __init__(self):\r
+ super().__init__()\r
+ self.oclouds = []\r
+\r
+ def _add(self, ocloud: ocloud.Ocloud):\r
+ self.oclouds.append(ocloud)\r
+\r
+ def _get(self, ocloudid) -> ocloud.Ocloud:\r
+ filtered = [o for o in self.oclouds if o.id == ocloudid]\r
+ return filtered.pop()\r
+\r
+ def _list(self) -> List[ocloud.Ocloud]:\r
+ return [x for x in self.oclouds]\r
+\r
+ def _update(self, ocloud: ocloud.Ocloud):\r
+ filtered = [o for o in self.oclouds if o.id == ocloud.id]\r
+ assert len(filtered) == 1\r
+ ocloud1 = filtered.pop()\r
+ ocloud1.update_by(ocloud)\r
+\r
+def test_probe_new_ocloud():\r
+ fakeRepo = FakeOcloudRepo()\r
+ fakeClient = FakeOcloudClient()\r
+ ocloudwatcher = OcloudWather(fakeClient, fakeRepo)\r
+ ocloudwatcher.probe()\r
+ assert len(fakeRepo.oclouds) == 1\r
+ assert fakeRepo.oclouds[0].name == "stx1"\r