|
10 | 10 | import os
|
11 | 11 | import re
|
12 | 12 | import subprocess
|
13 |
| -from typing import Any, Dict, Generator, Iterable, List, Optional |
| 13 | +from typing import Generator, Iterable, List, Optional |
14 | 14 |
|
| 15 | +from juju.client.client import FullStatus as JujuStatus |
15 | 16 | from jockey.cache import load_cache, new_cache_context, update_cache
|
16 | 17 | from jockey.log import configure_logging
|
17 | 18 |
|
18 | 19 |
|
19 | 20 | logger = logging.getLogger(__name__)
|
20 | 21 |
|
21 | 22 |
|
22 |
| -JujuStatus = Dict[str, Any] |
23 |
| - |
24 |
| - |
25 | 23 | class FilterMode(Enum):
|
26 | 24 | EQUALS = "="
|
27 | 25 | CONTAINS = "~"
|
@@ -301,7 +299,7 @@ def is_app_principal(status: JujuStatus, app_name: str) -> bool:
|
301 | 299 | is_principal (bool)
|
302 | 300 | Whether the indicated application is principal.
|
303 | 301 | """
|
304 |
| - return "subordinate-to" not in status["applications"][app_name] |
| 302 | + return status["applications"][app_name].get("subordinate-to") is None |
305 | 303 |
|
306 | 304 |
|
307 | 305 | def get_principal_unit_for_subordinate(status: JujuStatus, unit_name: str) -> str:
|
@@ -957,7 +955,7 @@ def get_juju_status(
|
957 | 955 | logger.debug("Loading local Juju status from %r", file)
|
958 | 956 | # print(dir(file))
|
959 | 957 | with open(file, "r") as f:
|
960 |
| - return json.loads(f.read()) |
| 958 | + return JujuStatus.from_json(f.read()) |
961 | 959 |
|
962 | 960 | # Get model name and build a CacheContext
|
963 | 961 | model_name = model_name or os.environ.get("JUJU_MODEL", None)
|
|
0 commit comments