mirror of
https://github.com/Sea-Haven-Industries/afterhours-shift-manager.git
synced 2026-09-30 09:03:11 +00:00
8a23d06a64
* ci(workflows): call org reusable CI and Fargate CD Local CI and the image deploy duplicated the org workflows and still required ci / ci. Pin the callers to those workflows and trust the reusable deploy ref. * test(ci): probe ruff with an undefined name * test(ci): remove the undefined-name ruff probe * ci: retrigger checks after removing the ruff probe * test(ci): probe ruff with an unused import * style: apply formatter * test(ci): remove the autofix probe --------- Co-authored-by: sea-haven-auto-fix[bot] <332630863+sea-haven-auto-fix[bot]@users.noreply.github.com>
101 lines
3.1 KiB
Python
101 lines
3.1 KiB
Python
"""copy_secrets.py writes Slack tokens into empty Terraform shells."""
|
|
|
|
import importlib.util
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from botocore.exceptions import ClientError
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
def _load():
|
|
spec = importlib.util.spec_from_file_location(
|
|
"copy_secrets", ROOT / "scripts" / "cutover" / "copy_secrets.py"
|
|
)
|
|
mod = importlib.util.module_from_spec(spec)
|
|
sys.modules["copy_secrets"] = mod
|
|
spec.loader.exec_module(mod)
|
|
return mod
|
|
|
|
|
|
mod = _load()
|
|
|
|
|
|
def _client_error(code: str) -> ClientError:
|
|
return ClientError({"Error": {"Code": code, "Message": code}}, "GetSecretValue")
|
|
|
|
|
|
class FakeSecrets:
|
|
def __init__(self, described, strings=None, get_errors=None):
|
|
self.described = set(described)
|
|
self.strings = dict(strings or {})
|
|
self.get_errors = dict(get_errors or {})
|
|
self.puts = []
|
|
|
|
def describe_secret(self, SecretId):
|
|
if SecretId not in self.described:
|
|
raise _client_error("ResourceNotFoundException")
|
|
return {"Name": SecretId}
|
|
|
|
def get_secret_value(self, SecretId):
|
|
if SecretId in self.get_errors:
|
|
raise _client_error(self.get_errors[SecretId])
|
|
if SecretId not in self.strings:
|
|
raise _client_error("ResourceNotFoundException")
|
|
return {"SecretString": self.strings[SecretId]}
|
|
|
|
def put_secret_value(self, SecretId, SecretString):
|
|
self.puts.append((SecretId, SecretString))
|
|
self.strings[SecretId] = SecretString
|
|
return {}
|
|
|
|
|
|
def test_execute_puts_into_empty_terraform_shells():
|
|
src = FakeSecrets(
|
|
described=mod.COPY,
|
|
strings={name: f"{name}-value\n" for name in mod.COPY},
|
|
)
|
|
dst = FakeSecrets(
|
|
described=mod.COPY + mod.VERIFY_ONLY,
|
|
strings={name: "already-copied" for name in mod.VERIFY_ONLY},
|
|
get_errors={name: "InvalidRequestException" for name in mod.COPY},
|
|
)
|
|
rc = mod.copy_secrets(src, dst, execute=True)
|
|
assert rc == 0
|
|
assert [name for name, _ in dst.puts] == list(mod.COPY)
|
|
assert all(
|
|
value.endswith("-value") and not value.endswith("\n") for _, value in dst.puts
|
|
)
|
|
|
|
|
|
def test_skip_populated_copy_targets_and_never_write_3cx():
|
|
src = FakeSecrets(
|
|
described=mod.COPY,
|
|
strings={name: "from-mgmt" for name in mod.COPY},
|
|
)
|
|
dst = FakeSecrets(
|
|
described=mod.COPY + mod.VERIFY_ONLY,
|
|
strings={
|
|
**{name: "prod-already" for name in mod.COPY},
|
|
**{name: "3cx-prod" for name in mod.VERIFY_ONLY},
|
|
},
|
|
)
|
|
rc = mod.copy_secrets(src, dst, execute=True)
|
|
assert rc == 0
|
|
assert dst.puts == []
|
|
|
|
|
|
def test_dry_run_does_not_put():
|
|
src = FakeSecrets(
|
|
described=mod.COPY,
|
|
strings={name: "from-mgmt" for name in mod.COPY},
|
|
)
|
|
dst = FakeSecrets(
|
|
described=mod.COPY + mod.VERIFY_ONLY,
|
|
strings={name: "3cx-prod" for name in mod.VERIFY_ONLY},
|
|
get_errors={name: "InvalidRequestException" for name in mod.COPY},
|
|
)
|
|
rc = mod.copy_secrets(src, dst, execute=False)
|
|
assert rc == 0
|
|
assert dst.puts == []
|