Code Coverage

Cobertura Coverage Report > osm_lcm >

osm_config.py

Trend

File Coverage summary

NameClassesLinesConditionals
osm_config.py
100%
1/1
100%
21/21
100%
0/0

Coverage Breakdown by Class

NameLinesConditionals
osm_config.py
100%
21/21
N/A

Source

osm_lcm/osm_config.py
1 # Copyright 2022 Canonical Ltd.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 #     Unless required by applicable law or agreed to in writing, software
10 #     distributed under the License is distributed on an "AS IS" BASIS,
11 #     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 #     See the License for the specific language governing permissions and
13 #     limitations under the License.
14
15 1 from typing import Any, Dict, List, Optional
16 1 from pydantic import BaseModel, validator
17
18
19 1 def _get_ip_from_service(service: Dict[str, Any]) -> List[str]:
20 1     return (
21         [service["cluster_ip"]]
22         if service["type"] == "ClusterIP"
23         else service["external_ip"]
24     )
25
26
27 1 class K8sConfigV0(BaseModel):
28 1     services: List[Dict]
29
30 1     @validator("services")
31 1     def parse_services(cls, services: Dict[str, Any]):
32 1         return {
33             service["name"]: {
34                 "type": service["type"],
35                 "ip": _get_ip_from_service(service),
36                 "ports": {
37                     port["name"]: {
38                         "port": port["port"],
39                         "protocol": port["protocol"],
40                     }
41                     for port in service["ports"]
42                 },
43             }
44             for service in services
45         }
46
47
48 1 class OsmConfigV0(BaseModel):
49 1     k8s: Optional[K8sConfigV0]
50
51
52 1 class OsmConfig(BaseModel):
53 1     v0: OsmConfigV0
54
55
56 1 class OsmConfigBuilder:
57 1     def __init__(self, k8s: Dict[str, Any] = {}) -> None:
58 1         self._k8s = k8s
59 1         self._configs = {}
60 1         if k8s:
61 1             self._configs["k8s"] = k8s
62
63 1     def build(self) -> Dict[str, Any]:
64 1         return OsmConfig(v0=OsmConfigV0(**self._configs)).dict()