gongzhiyong
111be3e435
The standalone prototype should be the root-level project shape for fengqun while preserving the existing planning documents already at the root. This keeps README, examples, tests, and the Python package directly discoverable without deleting the prior docs. Constraint: User clarified that swarm-minimal is the repository root, but other existing root files must remain. Rejected: Deleting existing root docs | They are part of the fengqun repository context and were explicitly protected. Confidence: high Scope-risk: narrow Directive: Keep secrets in ignored .env only; do not commit live credentials. Tested: python3 -B -m unittest discover -s tests; git diff --check; secret-pattern scan showed only placeholders/test values/task-id false positives. Not-tested: Remote web UI rendering after push.
94 lines
3.7 KiB
Python
94 lines
3.7 KiB
Python
import unittest
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
|
|
from swarm_minimal.azure_resources import azure_resource_plan
|
|
from swarm_minimal.config import BlobConfig, PostgresConfig, SwarmConfig, parse_redis_config
|
|
from swarm_minimal.core import InMemorySwarmStore, SwarmCoordinator, TaskStatus, default_agents
|
|
from swarm_minimal.local_env import load_env_file
|
|
|
|
|
|
class MinimalSwarmTest(unittest.TestCase):
|
|
def test_swarm_runs_four_shared_resources_to_convergence(self) -> None:
|
|
store = InMemorySwarmStore()
|
|
coordinator = SwarmCoordinator(store=store, agents=default_agents())
|
|
|
|
run_id = coordinator.submit_goal("ship the minimal swarm")
|
|
result = coordinator.run_until_converged(run_id)
|
|
|
|
self.assertEqual(result.completed_tasks, 3)
|
|
self.assertEqual(store.shared_state[f"run:{run_id}:status"], "converged")
|
|
self.assertTrue(all(task.status == TaskStatus.DONE for task in store.tasks.values()))
|
|
self.assertEqual(len(store.pheromones), 3)
|
|
self.assertEqual(store.convergence[run_id], result)
|
|
self.assertEqual(result.accepted_output, "Verify acceptance for: ship the minimal swarm")
|
|
|
|
def test_azure_plan_uses_pgsql_and_redis_without_nats_or_cosmos(self) -> None:
|
|
resources = azure_resource_plan()
|
|
text = " ".join(f"{item.key} {item.service} {item.purpose}" for item in resources)
|
|
|
|
self.assertIn("PostgreSQL", text)
|
|
self.assertIn("Redis", text)
|
|
self.assertIn("Blob Storage", text)
|
|
self.assertIn("Kubernetes", text)
|
|
self.assertNotIn("NATS", text)
|
|
self.assertNotIn("Cosmos", text)
|
|
|
|
def test_redis_connection_string_is_parsed_without_leaking_secret(self) -> None:
|
|
config = parse_redis_config("cache.example.net:6380,password=secret-value,ssl=True,abortConnect=False")
|
|
|
|
self.assertEqual(config.host, "cache.example.net")
|
|
self.assertEqual(config.port, 6380)
|
|
self.assertTrue(config.ssl)
|
|
self.assertEqual(config.password, "secret-value")
|
|
|
|
def test_config_summary_redacts_secrets(self) -> None:
|
|
redis = parse_redis_config("cache.example.net:6380,password=secret-value,ssl=True")
|
|
summary = SwarmConfig(
|
|
postgres=PostgresConfig(
|
|
host="pg.example.net",
|
|
port=5432,
|
|
database="swar",
|
|
user="azure",
|
|
password="pg-secret",
|
|
),
|
|
redis=redis,
|
|
blob=BlobConfig(
|
|
connection_string=(
|
|
"DefaultEndpointsProtocol=https;AccountName=acct;"
|
|
"AccountKey=blob-secret;EndpointSuffix=core.windows.net"
|
|
),
|
|
container="swarm-artifacts",
|
|
),
|
|
).redacted_summary()
|
|
|
|
text = str(summary)
|
|
self.assertIn("<redacted>", text)
|
|
self.assertNotIn("pg-secret", text)
|
|
self.assertNotIn("secret-value", text)
|
|
self.assertNotIn("blob-secret", text)
|
|
|
|
def test_local_env_file_loads_fake_values(self) -> None:
|
|
with TemporaryDirectory() as tmp:
|
|
path = Path(tmp) / ".env"
|
|
path.write_text(
|
|
"\n".join(
|
|
[
|
|
"FAKE_SWARM_ENV_ALPHA=one",
|
|
"FAKE_SWARM_ENV_BETA='two words'",
|
|
"# ignored",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
loaded = load_env_file(path, override=True)
|
|
|
|
self.assertEqual(loaded, 2)
|
|
self.assertEqual(__import__("os").environ["FAKE_SWARM_ENV_ALPHA"], "one")
|
|
self.assertEqual(__import__("os").environ["FAKE_SWARM_ENV_BETA"], "two words")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|