1 # Copyright (C) 2021 Wind River Systems, Inc.
\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
7 # http://www.apache.org/licenses/LICENSE-2.0
\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
15 from datetime import datetime
\r
17 from typing import List
\r
18 from o2ims.service.client.base_client import BaseClient
\r
20 from o2ims.domain import ocloud
\r
21 from o2ims import config
\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
27 class FakeOcloudClient(BaseClient):
\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
38 def _get(self, id) -> ocloudModel.StxGenericModel:
\r
39 return self.fakeCloud
\r
42 return [self.fakeCloud]
\r
44 class FakeOcloudRepo(OcloudRepository):
\r
49 def _add(self, ocloud: ocloud.Ocloud):
\r
50 self.oclouds.append(ocloud)
\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
56 def _list(self) -> List[ocloud.Ocloud]:
\r
57 return [x for x in self.oclouds]
\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
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