feat(config): add CLI smoke command and dependencies

Add central-cli with config-store-check command that:
- Connects via bootstrap config
- Lists adapters from config store
- Verifies crypto round-trip

Updates pyproject.toml with new dependencies:
- pydantic-settings>=2.7.0
- cryptography>=44.0.0

New entry points:
- central-migrate
- central-cli

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Ubuntu 2026-05-15 23:08:03 +00:00
commit 3e392cad81
2 changed files with 79 additions and 0 deletions

View file

@ -15,14 +15,18 @@ dependencies = [
"aiolimiter>=1.2.1", "aiolimiter>=1.2.1",
"asyncpg>=0.31.0", "asyncpg>=0.31.0",
"cloudevents>=2.0.0", "cloudevents>=2.0.0",
"cryptography>=44.0.0",
"nats-py>=2.14.0", "nats-py>=2.14.0",
"pydantic>=2,<3", "pydantic>=2,<3",
"pydantic-settings>=2.7.0",
"tenacity>=9.1.4", "tenacity>=9.1.4",
] ]
[project.scripts] [project.scripts]
central-supervisor = "central.supervisor:main" central-supervisor = "central.supervisor:main"
central-archive = "central.archive:main" central-archive = "central.archive:main"
central-migrate = "central.migrate:main"
central-cli = "central.cli:main"
[tool.hatch.build.targets.wheel] [tool.hatch.build.targets.wheel]
packages = ["src/central"] packages = ["src/central"]

75
src/central/cli.py Normal file
View file

@ -0,0 +1,75 @@
"""Central CLI commands."""
import argparse
import asyncio
import sys
async def config_store_check() -> int:
"""Smoke test for config store connectivity.
Connects via bootstrap_config, lists adapters, and verifies crypto.
Returns 0 on success, 1 on failure.
"""
from central.bootstrap_config import get_settings
from central.config_store import ConfigStore
from central.crypto import decrypt, encrypt
settings = get_settings()
print(f"Connecting to: {settings.db_dsn.split('@')[1]}") # Hide password
try:
store = await ConfigStore.create(settings.db_dsn)
except Exception as e:
print(f"ERROR: Failed to connect to database: {e}")
return 1
try:
# List adapters
adapters = await store.list_adapters()
print(f"\nAdapters ({len(adapters)}):")
for adapter in adapters:
print(f" - {adapter.name}: enabled={adapter.enabled}, cadence_s={adapter.cadence_s}")
print(f" settings: {adapter.settings}")
# Test crypto
test_plaintext = b"config_store_check_test"
try:
ciphertext = encrypt(test_plaintext)
decrypted = decrypt(ciphertext)
if decrypted == test_plaintext:
print("\ncrypto: ok")
else:
print("\ncrypto: FAILED (round-trip mismatch)")
return 1
except Exception as e:
print(f"\ncrypto: FAILED ({e})")
return 1
print("\nAll checks passed.")
return 0
finally:
await store.close()
def main_config_store_check() -> None:
"""Entry point for central-cli config-store-check."""
sys.exit(asyncio.run(config_store_check()))
def main() -> None:
"""Main CLI entry point."""
parser = argparse.ArgumentParser(description="Central CLI")
subparsers = parser.add_subparsers(dest="command", required=True)
subparsers.add_parser("config-store-check", help="Test config store connectivity")
args = parser.parse_args()
if args.command == "config-store-check":
main_config_store_check()
if __name__ == "__main__":
main()