[{'Text': '"""Tests for GBS tools (get_status, add_statistic, create_source).\n\nThese mock mongodb_utils to test validation, formatting, and error paths\nwithout requiring a real MongoDB connection.\n"""\n\nimport json\nfrom unittest.mock import patch\n\nimport pytest\n\nfrom bson import ObjectId\n\n\n@pytest.fixture\ndef gbs_env():\n    """Ensure PT_RESEARCH_DATABASE_URI looks set so tools register."""\n    with patch.dict("os.environ", {"PT_RESEARCH_DATABASE_URI": "mongodb://localhost:27017/t"}, clear=False):\n        # Reload gbs_tools to trigger conditional import\n        import importlib\n        import point_topic_mcp.tools.gbs_tools as gbs\n        importlib.reload(gbs)\n        yield gbs\n\n\ndef test_gbs_get_status_country_level(gbs_env):\n    """Country-level: returns coverage rollup with operator lists."""\n    gbs = gbs_env\n    op_id_a = ObjectId()\n    op_id_b = ObjectId()\n    mock_docs = [\n        {"_id": {"$oid": str(op_id_a)}, "name": "Alpha", "techs": ["FTTP"], "statCount": 5, "stats": [{"state": 3}] * 5},\n        {"_id": {"$oid": str(op_id_b)}, "name": "Beta", "techs": ["DSL"], "statCount": 0, "stats": []},\n    ]\n    mock_return = json.dumps({"currentPeriod": "2025Q1", "targetPeriod": "2025Q1", "docs": mock_docs})\n\n    with patch("point_topic_mcp.core.mongodb_utils._run_mongosh", return_value=mock_return):\n        result = gbs.gbs_get_status(country="UK")\n        assert "GBS STATUS: UK" in result\n        assert "period 2025Q1" in result\n        assert "2 total" in result\n        assert "1 with data" in result\n        assert "1 without data" in result\n        assert "Alpha" in result\n        assert "Beta" in result\n        assert "50.0%" in result  # 1/2 coverage\n\n\ndef test_gbs_get_status_operator_not_found(gbs_env):\n    """Operator-level: returns error when operator doesn\'t exist."""\n    gbs = gbs_env\n    mock_return = json.dumps({"currentPeriod": "2025Q1", "targetPeriod": "2025Q1", "docs": []})\n\n    with patch("point_topic_mcp.core.mongodb_utils._run_mongosh", return_value=mock_return):\n        result = gbs.gbs_get_status(country="UK", operator="NoSuchOp")\n        assert "not found" in result\n        assert "NoSuchOp" in result\n\n\ndef test_gbs_get_status_operator_level(gbs_env):\n    """Operator-level: returns records, gaps, state breakdown, admin URL."""\n    gbs = gbs_env\n    op_id = ObjectId()\n    mock_stats = [\n        {"type": "broadband", "tech": "FTTP", "channel": "Infrastructure",\n         "domain": "Residential", "subscribers": 100000, "state": 1},\n    ]\n    mock_docs = [{\n        "_id": {"$oid": str(op_id)},\n        "name": "TestOp",\n        "techs": ["FTTP", "FTTC"],\n        "statCount": 1,\n        "stats": mock_stats,\n    }]\n    mock_return = json.dumps({"currentPeriod": "2025Q1", "targetPeriod": "2025Q1", "docs": mock_docs})\n\n    with patch("point_topic_mcp.core.mongodb_utils._run_mongosh", return_value=mock_return):\n        result = gbs.gbs_get_status(country="UK", operator="TestOp")\n        assert "GBS STATUS: TestOp (UK)" in result\n        assert "ID:        " + str(op_id) in result\n        assert "Records:   1" in result\n        assert "Filled: 1 combos" in result\n        assert "Gaps" in result\n        assert "pending: 1" in result\n        assert "broadband: 1" in result\n        assert "Admin:" in result\n        assert str(op_id) in result\n        # Should show the record\n        assert "100,000" in result\n        assert "FTTP" in result\n        assert "Infrastructure" in result\n        assert "Residential" in result\n\n\ndef test_gbs_get_status_operator_staleness(gbs_env):\n    """Operator-level: works with explicit period; any period is accepted."""\n    gbs = gbs_env\n    op_id = ObjectId()\n    mock_docs = [{\n        "_id": {"$oid": str(op_id)},\n        "name": "TestOp",\n        "techs": ["FTTP"],\n        "statCount": 0,\n        "stats": [],\n    }]\n    # Period specified → should be used as targetPeriod\n    mock_return = json.dumps({"currentPeriod": "2025Q3", "targetPeriod": "2025Q1", "docs": mock_docs})\n\n    with patch("point_topic_mcp.core.mongodb_utils._run_mongosh", return_value=mock_return):\n        result = gbs.gbs_get_status(country="UK", operator="TestOp", period="2025Q1")\n        assert "2025Q1" in result\n        assert "Records:   0" in result\n\n\ndef test_gbs_add_statistic_validation(gbs_env):\n    """gbs_add_statistic returns errors for invalid inputs."""\n    gbs = gbs_env\n\n    result = gbs.gbs_add_statistic(\n        type="invalid",\n        operator_id="000000000000000000000001",\n        period="2025Q1",\n        tech="FTTP",\n        channel="Wrong",\n        domain="Residential",\n        subscribers=1000,\n    )\n    assert "Invalid" in result\n\n    result = gbs.gbs_add_statistic(\n        type="broadband",\n        operator_id="bad",\n        period="2025Q1",\n        tech="FTTP",\n        channel="Infrastructure",\n        domain="Residential",\n        subscribers=1000,\n    )\n    assert "Invalid operator_id" in result\n\n    result = gbs.gbs_add_statistic(\n        type="broadband",\n        operator_id="000000000000000000000001",\n        period="bad",\n        tech="FTTP",\n        channel="Infrastructure",\n        domain="Residential",\n        subscribers=1000,\n    )\n    assert "Invalid period" in result\n\n    result = gbs.gbs_add_statistic(\n        type="broadband",\n        operator_id="000000000000000000000001",\n        period="2025Q1",\n        tech="FTTP",\n        channel="Infrastructure",\n        domain="Residential",\n        subscribers=-1,\n    )\n    assert "Invalid subscribers" in result\n\n\ndef test_gbs_add_statistic_success(gbs_env):\n    """gbs_add_statistic inserts a pending record with admin URL."""\n    gbs = gbs_env\n    op_id = str(ObjectId())\n    mock_response = json.dumps({\n        "ok": True,\n        "insertedId": op_id,\n        "operator": "TestOp",\n        "country": "UK",\n        "tech": "FTTP",\n    })\n\n    with patch("point_topic_mcp.core.mongodb_utils.insert_statistic_safe", return_value=mock_response):\n        result = gbs.gbs_add_statistic(\n            type="broadband",\n            operator_id=op_id,\n            period="2025Q3",\n            tech="FTTP",\n            channel="Infrastructure",\n            domain="Residential",\n            subscribers=50000,\n            created_by="test-agent",\n        )\n        assert "Created statistic" in result\n        assert "state: pending" in result\n        assert "TestOp" in result\n        assert "50,000" in result\n        assert "Admin:" in result\n        assert op_id in result\n\n\ndef test_gbs_add_statistic_duplicate_rejected(gbs_env):\n    """gbs_add_statistic rejects duplicate (operator, period, type, tech, channel, domain)."""\n    gbs = gbs_env\n    op_id = str(ObjectId())\n    existing_id = str(ObjectId())\n    mock_response = json.dumps({\n        "ok": False,\n        "error": f"Duplicate: statistic already exists for broadband/FTTP/Infrastructure/Residential in 2025Q3 (id={existing_id}, state=1)",\n    })\n\n    with patch("point_topic_mcp.core.mongodb_utils.insert_statistic_safe", return_value=mock_response):\n        result = gbs.gbs_add_statistic(\n            type="broadband",\n            operator_id=op_id,\n            period="2025Q3",\n            tech="FTTP",\n            channel="Infrastructure",\n            domain="Residential",\n            subscribers=50000,\n        )\n        assert "Duplicate" in result\n        assert existing_id in result\n\n\ndef test_gbs_create_source_validation(gbs_env):\n    """gbs_create_source returns errors for invalid inputs."""\n    gbs = gbs_env\n\n    result = gbs.gbs_create_source(\n        operator_id="bad",\n        year=2025,\n        quarter=1,\n        type="report",\n    )\n    assert "Invalid operator_id" in result\n\n    result = gbs.gbs_create_source(\n        operator_id="000000000000000000000001",\n        year=2025,\n        quarter=5,\n        type="report",\n    )\n    assert "Invalid quarter" in result\n\n    result = gbs.gbs_create_source(\n        operator_id="000000000000000000000001",\n        year=2025,\n        quarter=1,\n        type="",\n    )\n    assert "Invalid type" in result\n\n    result = gbs.gbs_create_source(\n        operator_id="000000000000000000000001",\n        year=2025,\n        quarter=1,\n        type="report",\n        url="",\n    )\n    assert "At least one" in result\n\n\ndef test_gbs_create_source_success(gbs_env):\n    """gbs_create_source inserts a source record."""\n    gbs = gbs_env\n    op_id = str(ObjectId())\n    mock_op = {"_id": {"$oid": op_id}, "name": "TestOp"}\n    source_id = str(ObjectId())\n\n    with (\n        patch("point_topic_mcp.core.mongodb_utils.find_one_operator_by_id", return_value=mock_op),\n        patch("point_topic_mcp.core.mongodb_utils.find_sources", return_value=[]),\n        patch("point_topic_mcp.core.mongodb_utils.insert_one_source", return_value=source_id),\n    ):\n        result = gbs.gbs_create_source(\n            operator_id=op_id,\n            year=2025,\n            quarter=3,\n            type="report",\n            url="https://example.com/report",\n        )\n        assert "Created source" in result\n        assert source_id in result\n        assert "TestOp" in result\n        assert "2025Q3" in result\n\n\ndef test_gbs_create_source_duplicate_rejected(gbs_env):\n    """gbs_create_source rejects exact duplicates."""\n    gbs = gbs_env\n    op_id = str(ObjectId())\n    mock_op = {"_id": {"$oid": op_id}, "name": "TestOp"}\n    existing_id = str(ObjectId())\n    mock_existing = [{\n        "_id": {"$oid": existing_id},\n        "type": "report",\n        "url": "https://example.com/report",\n        "fileUrl": None,\n    }]\n\n    with (\n        patch("point_topic_mcp.core.mongodb_utils.find_one_operator_by_id", return_value=mock_op),\n        patch("point_topic_mcp.core.mongodb_utils.find_sources", return_value=mock_existing),\n    ):\n        result = gbs.gbs_create_source(\n            operator_id=op_id,\n            year=2025,\n            quarter=3,\n            type="report",\n            url="https://example.com/report",\n        )\n        assert "Duplicate" in result\n        assert existing_id in result\n\n\ndef test_discovery_uses_prefix(gbs_env):\n    """Tool auto-discovery prefixes module name."""\n    import point_topic_mcp.tools as tools\n    all_tools = tools.get_all_tools()\n    names = [n for n, _ in all_tools]\n    assert "gbs_tools_gbs_get_status" in names\n    assert "gbs_tools_gbs_add_statistic" in names\n    assert "gbs_tools_gbs_create_source" in names\n    # Old names should be gone\n    old_names = ["gbs_tools_list_operators", "gbs_tools_get_gbs_status",\n                  "gbs_tools_add_statistic", "gbs_tools_create_source"]\n    for old in old_names:\n        assert old not in names, f"Old tool \'{old}\' should be removed"\n'}]