""" 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"