yadunut/clan-core
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

refactor secrets & facts -> secret_facts & public_facts

Browse Source
This commit is contained in:
lassulus
2024-03-23 05:05:31 +01:00
parent 6638d7a5cf
commit 8d34c51df2
26 changed files with 116 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
pkgs/clan-cli/clan_cli/facts/check.py
View File

@@ -7,32 +7,55 @@ from ..machines.machines import Machine
log = logging.getLogger(__name__)
def check_facts(machine: Machine) -> bool:
facts_module = importlib.import_module(machine.facts_module)
fact_store = facts_module.FactStore(machine=machine)
def check_secrets(machine: Machine, service: None | str = None) -> bool:
secret_facts_module = importlib.import_module(machine.secret_facts_module)
secret_facts_store = secret_facts_module.SecretStore(machine=machine)
public_facts_module = importlib.import_module(machine.public_facts_module)
public_facts_store = public_facts_module.FactStore(machine=machine)
existing_facts = fact_store.get_all()
missing_facts = []
for service in machine.secrets_data:
for fact in machine.secrets_data[service]["facts"]:
if fact not in existing_facts.get(service, {}):
log.info(f"Fact {fact} for service {service} is missing")
missing_facts.append((service, fact))
missing_secret_facts = []
missing_public_facts = []
if service:
services = [service]
else:
services = list(machine.secrets_data.keys())
for service in services:
for secret_fact in machine.secrets_data[service]["secrets"]:
if isinstance(secret_fact, str):
secret_name = secret_fact
else:
secret_name = secret_fact["name"]
if not secret_facts_store.exists(service, secret_name):
log.info(f"Secret fact {secret_fact} for service {service} is missing")
missing_secret_facts.append((service, secret_name))
if missing_facts:
for public_fact in machine.secrets_data[service]["facts"]:
if not public_facts_store.exists(service, public_fact):
log.info(f"public Fact {public_fact} for service {service} is missing")
missing_public_facts.append((service, public_fact))
log.debug(f"missing_secret_facts: {missing_secret_facts}")
log.debug(f"missing_public_facts: {missing_public_facts}")
if missing_secret_facts or missing_public_facts:
return False
return True
def check_command(args: argparse.Namespace) -> None:
machine = Machine(name=args.machine, flake=args.flake)
if check_facts(machine):
print("All facts are present")
machine = Machine(
name=args.machine,
flake=args.flake,
)
check_secrets(machine, service=args.service)
def register_check_parser(parser: argparse.ArgumentParser) -> None:
parser.add_argument(
"machine",
help="The machine to check facts for",
help="The machine to check secrets for",
)
parser.add_argument(
"--service",
help="the service to check",
)
parser.set_defaults(func=check_command)