""" Tests for Agent Instance mechanism. Tests cover: 1. _deep_merge() — dict merging semantics 2. create_instance() — directory and file creation 3. load_instance() — merge agent template + instance overrides 4. load_instance_from_dirs() — two-dir loading 5. Multiple instances from same template """ import json import os import shutil import tempfile import pytest import yaml from agentpaas.engine.instance import ( _deep_merge, create_instance, load_instance, load_instance_from_dirs, ) @pytest.fixture def tmpdir(): d = tempfile.mkdtemp(prefix="test_instance_") yield d shutil.rmtree(d, ignore_errors=True) def _write_yaml(path, data): os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w") as f: yaml.dump(data, f, allow_unicode=True) # ============================================================ # 1. _deep_merge # ============================================================ class TestDeepMerge: def test_simple_override(self): base = {"a": 1, "b": 2} over = {"b": 3, "c": 4} assert _deep_merge(base, over) == {"a": 1, "b": 3, "c": 4} def test_nested_merge(self): base = {"model": {"name": "qwen", "temperature": 0.3}} over = {"model": {"name": "gpt-4"}} result = _deep_merge(base, over) assert result["model"]["name"] == "gpt-4" assert result["model"]["temperature"] == 0.3 # preserved def test_deep_nested(self): base = {"a": {"b": {"c": 1, "d": 2}}} over = {"a": {"b": {"c": 99}}} result = _deep_merge(base, over) assert result["a"]["b"]["c"] == 99 assert result["a"]["b"]["d"] == 2 def test_no_mutation(self): base = {"x": {"y": 1}} over = {"x": {"y": 2}} result = _deep_merge(base, over) assert base["x"]["y"] == 1 # base unchanged # ============================================================ # 2. create_instance # ============================================================ class TestCreateInstance: def test_creates_directories(self, tmpdir): inst_dir = os.path.join(tmpdir, "maritime") create_instance(inst_dir, "qaagent67wiki", name="Maritime Wiki") assert os.path.isdir(os.path.join(inst_dir, "wiki")) assert os.path.isdir(os.path.join(inst_dir, "knowledge", "raw")) assert os.path.isdir(os.path.join(inst_dir, "knowledge", "processed")) assert os.path.isdir(os.path.join(inst_dir, "workspace")) def test_creates_instance_yml(self, tmpdir): inst_dir = os.path.join(tmpdir, "maritime") path = create_instance(inst_dir, "qaagent67wiki", name="Maritime Wiki") assert os.path.isfile(path) with open(path) as f: cfg = yaml.safe_load(f) assert cfg["agent"] == "qaagent67wiki" assert cfg["name"] == "Maritime Wiki" assert "knowledge" in cfg assert "wiki" in cfg def test_auto_paths(self, tmpdir): inst_dir = os.path.join(tmpdir, "finance") create_instance(inst_dir, "qaagent67wiki") with open(os.path.join(inst_dir, "instance.yml")) as f: cfg = yaml.safe_load(f) # Paths should point to instance_dir assert inst_dir in cfg["knowledge"]["baseDir"] assert inst_dir in cfg["wiki"]["dir"] # ============================================================ # 3. load_instance # ============================================================ class TestLoadInstance: def test_merge_template_and_instance(self, tmpdir): # Create agent template agent_dir = os.path.join(tmpdir, "agentexample", "myagent") _write_yaml(os.path.join(agent_dir, "agent-config.yml"), { "type": "react", "systemPrompt": "You are helpful", "model": {"name": "qwen", "temperature": 0.3}, "knowledge": {"baseDir": "./knowledge"}, }) # Create instance inst_dir = os.path.join(tmpdir, "instances", "domain1") os.makedirs(inst_dir, exist_ok=True) _write_yaml(os.path.join(inst_dir, "instance.yml"), { "agent": agent_dir, "name": "Domain 1", "knowledge": {"baseDir": "/data/domain1/knowledge"}, "model": {"name": "gpt-4"}, }) config = load_instance(os.path.join(inst_dir, "instance.yml")) # Template fields preserved assert config["type"] == "react" assert config["systemPrompt"] == "You are helpful" # Instance overrides applied assert config["knowledge"]["baseDir"] == "/data/domain1/knowledge" assert config["model"]["name"] == "gpt-4" # Nested merge: temperature preserved from template assert config["model"]["temperature"] == 0.3 # Metadata injected assert config["_agent_dir"] == agent_dir assert config["_instance_dir"] == inst_dir assert config["_instance_name"] == "Domain 1" # ============================================================ # 4. load_instance_from_dirs # ============================================================ class TestLoadInstanceFromDirs: def test_with_instance_yml(self, tmpdir): agent_dir = os.path.join(tmpdir, "agent") inst_dir = os.path.join(tmpdir, "instance") _write_yaml(os.path.join(agent_dir, "agent-config.yml"), { "type": "simple", "systemPrompt": "base", "knowledge": {"baseDir": "./kb"}, }) os.makedirs(inst_dir, exist_ok=True) _write_yaml(os.path.join(inst_dir, "instance.yml"), { "agent": agent_dir, "knowledge": {"baseDir": "/data/prod/kb"}, }) config = load_instance_from_dirs(agent_dir, inst_dir) assert config["knowledge"]["baseDir"] == "/data/prod/kb" assert config["systemPrompt"] == "base" def test_without_instance_yml(self, tmpdir): agent_dir = os.path.join(tmpdir, "agent") inst_dir = os.path.join(tmpdir, "instance") _write_yaml(os.path.join(agent_dir, "agent-config.yml"), { "type": "simple", "knowledge": {"baseDir": "./default"}, }) os.makedirs(inst_dir, exist_ok=True) config = load_instance_from_dirs(agent_dir, inst_dir) assert config["knowledge"]["baseDir"] == "./default" assert config["_instance_dir"] == os.path.abspath(inst_dir) # ============================================================ # 5. Multiple Instances from Same Template # ============================================================ class TestMultipleInstances: def test_two_instances_share_template(self, tmpdir): agent_dir = os.path.join(tmpdir, "template") _write_yaml(os.path.join(agent_dir, "agent-config.yml"), { "type": "react", "systemPrompt": "Wiki agent", "knowledge": {"baseDir": "./default"}, }) # Instance A: maritime inst_a = os.path.join(tmpdir, "maritime") os.makedirs(inst_a, exist_ok=True) _write_yaml(os.path.join(inst_a, "instance.yml"), { "agent": agent_dir, "name": "Maritime", "knowledge": {"baseDir": "/data/maritime"}, }) # Instance B: medical inst_b = os.path.join(tmpdir, "medical") os.makedirs(inst_b, exist_ok=True) _write_yaml(os.path.join(inst_b, "instance.yml"), { "agent": agent_dir, "name": "Medical", "knowledge": {"baseDir": "/data/medical"}, }) cfg_a = load_instance(os.path.join(inst_a, "instance.yml")) cfg_b = load_instance(os.path.join(inst_b, "instance.yml")) # Same template assert cfg_a["type"] == cfg_b["type"] == "react" assert cfg_a["systemPrompt"] == cfg_b["systemPrompt"] # Different data dirs assert cfg_a["knowledge"]["baseDir"] == "/data/maritime" assert cfg_b["knowledge"]["baseDir"] == "/data/medical" assert cfg_a["_instance_name"] == "Maritime" assert cfg_b["_instance_name"] == "Medical" # ============================================================ # 6. DB persistence (audit #36) # ============================================================ # # Up to this point the test suite covered only the helper layer # (_deep_merge, create_instance, load_instance). The new # `agents.agent_template` + `agents.instance_dir` DB columns — the # load-bearing piece of the documented multi-tenant template-vs-instance # model — were entirely unexercised. A typo in the INSERT or in the # round-tripping logic could ship to production with green CI. # # These tests use a fresh `:memory:` SQLite (same trick as # test_api_endpoints.py + test_authenticate.py) and verify: # - the two new columns round-trip correctly through INSERT + SELECT # - two agents sharing one template but pointing at distinct # instance_dirs do NOT collide on workspace paths class TestInstanceDBPersistence: @pytest.fixture def db(self): import os os.environ.setdefault("AGENTPAAS_DATABASE_URL", "sqlite:///:memory:") from agentpaas.db.models import Database import agentpaas.db.session as _session_mod prev = _session_mod._db _session_mod._db = Database("sqlite:///:memory:") try: yield _session_mod._db finally: _session_mod._db = prev def _insert_agent(self, db, agent_id: str, tenant_id: str, name: str, agent_template: str, instance_dir: str) -> None: from agentpaas.db.models import now_utc now = now_utc() db.execute( "INSERT INTO agents " "(id, tenant_id, name, description, agent_template, instance_dir, " " status, created_at, updated_at) " "VALUES (?, ?, ?, '', ?, ?, 'active', ?, ?)", (agent_id, tenant_id, name, agent_template, instance_dir, now, now), ) db.commit() def test_agent_template_and_instance_dir_round_trip(self, db, tmpdir): """INSERT → SELECT must preserve both columns. Audit #36 noted these two columns are the actual storage for the template-vs-instance model; a silent schema mismatch (e.g. column renamed but old migration left in place) would corrupt every tenant's agent records.""" from agentpaas.db.models import gen_id tid = "tn_test" # Seed tenant so the FK is satisfied (even though SQLite doesn't # enforce it by default, leaving it unset is sloppy and confuses # later schema introspection). db.execute( "INSERT INTO tenants (id, name, plan, status, created_at) " "VALUES (?, 'test', 'free', 'active', '2026-01-01')", (tid,), ) agent_dir = os.path.join(tmpdir, "template-agent") inst_dir = os.path.join(tmpdir, "inst-A") os.makedirs(agent_dir, exist_ok=True) os.makedirs(inst_dir, exist_ok=True) aid = gen_id("ag_") self._insert_agent(db, aid, tid, "wiki-A", agent_template=agent_dir, instance_dir=inst_dir) row = db.fetchone( "SELECT agent_template, instance_dir FROM agents WHERE id = ?", (aid,), ) assert row is not None assert row["agent_template"] == agent_dir assert row["instance_dir"] == inst_dir def test_two_instances_one_template_workspace_isolated(self, db, tmpdir): """The documented multi-tenant pattern: one template (agent_dir), N instances each with its own `instance_dir`. After dispatch, the workspace path must point at the instance_dir, NEVER at the shared template — otherwise two instances stomp each other. We don't run a full PaaS dispatch here (would need the whole Runtime stack); we verify the contract at the layer the audit flagged — that the DB layer correctly disambiguates the two agents AND that load_instance reproduces distinct workspace roots for them.""" from agentpaas.db.models import gen_id tid = "tn_test" db.execute( "INSERT INTO tenants (id, name, plan, status, created_at) " "VALUES (?, 'test', 'free', 'active', '2026-01-01')", (tid,), ) # Shared template: one agent-config.yml that both instances inherit # from. (Loader looks for `agent-config.yml`, not `agent.yml`; see # agentpaas/engine/instance.py:82.) template_dir = os.path.join(tmpdir, "template") os.makedirs(template_dir, exist_ok=True) _write_yaml(os.path.join(template_dir, "agent-config.yml"), { "type": "react", "systemPrompt": "Shared template", "knowledge": {"baseDir": "./default"}, }) # Two instance dirs, each with its own instance.yml override. inst_a_dir = os.path.join(tmpdir, "inst-A") inst_b_dir = os.path.join(tmpdir, "inst-B") for d, label in ((inst_a_dir, "A"), (inst_b_dir, "B")): os.makedirs(d, exist_ok=True) _write_yaml(os.path.join(d, "instance.yml"), { "agent": template_dir, "name": f"Inst{label}", "knowledge": {"baseDir": f"/data/{label.lower()}"}, }) # Two agent rows: same template, different instance_dirs. aid_a = gen_id("ag_") aid_b = gen_id("ag_") self._insert_agent(db, aid_a, tid, "wiki-A", agent_template=template_dir, instance_dir=inst_a_dir) self._insert_agent(db, aid_b, tid, "wiki-B", agent_template=template_dir, instance_dir=inst_b_dir) # DB layer: distinct instance_dirs preserved. a = db.fetchone("SELECT instance_dir FROM agents WHERE id = ?", (aid_a,)) b = db.fetchone("SELECT instance_dir FROM agents WHERE id = ?", (aid_b,)) assert a["instance_dir"] != b["instance_dir"] # Config layer: load_instance for each yields distinct # _instance_dir, so any downstream workspace_path computation # (whether {instance_dir}/workspace/run_X or otherwise) will # also be distinct. cfg_a = load_instance(os.path.join(inst_a_dir, "instance.yml")) cfg_b = load_instance(os.path.join(inst_b_dir, "instance.yml")) assert cfg_a["_instance_dir"] != cfg_b["_instance_dir"] assert cfg_a["knowledge"]["baseDir"] != cfg_b["knowledge"]["baseDir"] # Both still inherit the template's prompt. assert cfg_a["systemPrompt"] == cfg_b["systemPrompt"] == "Shared template" def test_agent_template_can_be_null_for_legacy_agents(self, db, tmpdir): """Legacy agents (created before the template/instance feature) will have NULL `agent_template` and NULL `instance_dir`. The DB column must allow that without erroring; SELECT must return None for those fields.""" from agentpaas.db.models import gen_id tid = "tn_test" db.execute( "INSERT INTO tenants (id, name, plan, status, created_at) " "VALUES (?, 'test', 'free', 'active', '2026-01-01')", (tid,), ) aid = gen_id("ag_") # Explicitly INSERT without setting agent_template / instance_dir # (use a minimal column list to skip them). db.execute( "INSERT INTO agents " "(id, tenant_id, name, status, created_at, updated_at) " "VALUES (?, ?, 'legacy-agent', 'active', " " '2026-01-01', '2026-01-01')", (aid, tid), ) db.commit() row = db.fetchone( "SELECT agent_template, instance_dir FROM agents WHERE id = ?", (aid,), ) assert row is not None assert row["agent_template"] in (None, "") assert row["instance_dir"] in (None, "")