mirror of
https://github.com/hanzoai/insights.git
synced 2026-10-30 16:15:31 +00:00
zeekay
40d120dbf7
Every first-party Python caller and test now names the mount the API actually answers on. Mechanical: URL literals only, and only where the /api/ was ours -- a preceding word character means it was a module path like insights/api/ or a third-party host like slack.com/api/, and those are untouched. Left saying /api/, each for a reason rather than for safety: - products/warehouse_sources/.../sources/** is a catalogue of vendor endpoints - services/llm-gateway deploys separately and is under a code freeze, so moving its two calls would break it during the skew - the mcp_store and Slack/Teams redirect_uris are registered with third parties and must byte-match what was registered - the Wizard and Raycast client-metadata paths are OAuth client_ids - /api/surveys and /api/feature_flag/local_evaluation are SDK aliases the ingress routes to the Rust fleets, held by released SDK versions Test URLconfs moved with the requests that call them, and the Datastore user-selection fixtures now carry the leading slash that request.path has.
294 lines
12 KiB
Python
294 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
import tempfile
|
|
import importlib.util
|
|
from pathlib import Path
|
|
from types import ModuleType
|
|
|
|
import unittest
|
|
|
|
SCRIPT_PATH = Path(__file__).with_name("snob_backend_test_selection_shadow.py")
|
|
|
|
|
|
def _load_selection_module() -> ModuleType:
|
|
spec = importlib.util.spec_from_file_location("snob_backend_test_selection_shadow", SCRIPT_PATH)
|
|
assert spec is not None
|
|
assert spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class TestSnobBackendTestSelectionShadow(unittest.TestCase):
|
|
def test_classifies_django_api_client_tests(self) -> None:
|
|
with tempfile.TemporaryDirectory() as root:
|
|
tmp_path = Path(root)
|
|
selection = _load_selection_module()
|
|
selection.REPO_ROOT = tmp_path
|
|
|
|
test_path = tmp_path / "products" / "feature_flags" / "backend" / "test" / "test_api.py"
|
|
test_path.parent.mkdir(parents=True)
|
|
test_path.write_text(
|
|
"\n".join(
|
|
[
|
|
"from rest_framework.test import APIClient",
|
|
"",
|
|
"def test_feature_flags_endpoint():",
|
|
" client = APIClient()",
|
|
" client.get('/v1/projects/1/feature_flags/')",
|
|
]
|
|
)
|
|
)
|
|
|
|
features = selection.classify_test_file("products/feature_flags/backend/test/test_api.py")
|
|
|
|
self.assertTrue(features.imports_api_client)
|
|
self.assertTrue(features.calls_http_client)
|
|
self.assertTrue(features.uses_api_url)
|
|
self.assertTrue(features.is_django_api_test)
|
|
self.assertIn("feature_flags", features.api_tokens)
|
|
|
|
def test_ast_selection_groups_product_api_client_tests(self) -> None:
|
|
with tempfile.TemporaryDirectory() as root:
|
|
tmp_path = Path(root)
|
|
selection = _load_selection_module()
|
|
selection.REPO_ROOT = tmp_path
|
|
|
|
test_path = tmp_path / "products" / "feature_flags" / "backend" / "test" / "test_api.py"
|
|
test_path.parent.mkdir(parents=True)
|
|
test_path.write_text(
|
|
"\n".join(
|
|
[
|
|
"def test_feature_flags_endpoint(api_client):",
|
|
" api_client.get('/v1/projects/1/feature_flags/')",
|
|
]
|
|
)
|
|
)
|
|
|
|
features_by_path = selection.classify_tests()
|
|
result = selection.ast_select_tests(
|
|
["products/feature_flags/backend/api/feature_flags.py"],
|
|
features_by_path,
|
|
)
|
|
|
|
self.assertIn("product_api_client:feature_flags", result.groups)
|
|
self.assertIn("product_api_route_tokens:feature_flags", result.groups)
|
|
self.assertIn("same_app:products/feature_flags/backend", result.groups)
|
|
self.assertEqual(["products/feature_flags/backend/test/test_api.py"], result.tests)
|
|
|
|
def test_ast_selection_matches_insights_api_test_by_filename(self) -> None:
|
|
selection = _load_selection_module()
|
|
|
|
result = selection.ast_select_tests(
|
|
["insights/api/project.py"],
|
|
{
|
|
"insights/api/test/test_project.py": selection.TestFeatures(
|
|
path="insights/api/test/test_project.py",
|
|
imports_api_client=True,
|
|
api_tokens=("project",),
|
|
),
|
|
"insights/api/test/test_user.py": selection.TestFeatures(
|
|
path="insights/api/test/test_user.py",
|
|
imports_api_client=True,
|
|
api_tokens=("user",),
|
|
),
|
|
},
|
|
)
|
|
|
|
self.assertIn("conventional_neighbors", result.groups)
|
|
self.assertIn("insights_api_route_tokens", result.groups)
|
|
# same-app fallback includes all tests under insights/api/
|
|
self.assertIn("same_app:insights/api", result.groups)
|
|
self.assertIn("insights/api/test/test_project.py", result.tests)
|
|
self.assertIn("insights/api/test/test_user.py", result.tests)
|
|
|
|
def test_snob_selection_filters_to_python_files(self) -> None:
|
|
selection = _load_selection_module()
|
|
with tempfile.TemporaryDirectory() as root:
|
|
selection.REPO_ROOT = Path(root)
|
|
selected_test = selection.REPO_ROOT / "insights" / "api" / "test" / "test_feature_flags.py"
|
|
selected_test.parent.mkdir(parents=True)
|
|
selected_test.write_text("def test_feature_flags():\n pass\n")
|
|
|
|
seen_changed_files: list[list[str]] = []
|
|
|
|
fake_snob = ModuleType("snob_lib")
|
|
|
|
def get_tests(changed_files: list[str]) -> set[str]:
|
|
seen_changed_files.append(changed_files)
|
|
return {str(selected_test)}
|
|
|
|
fake_snob.get_tests = get_tests # type: ignore[attr-defined]
|
|
previous_snob = sys.modules.get("snob_lib")
|
|
sys.modules["snob_lib"] = fake_snob
|
|
try:
|
|
result = selection.snob_select_tests(["insights/api/feature_flags.py", "frontend/src/index.ts"])
|
|
finally:
|
|
if previous_snob is None:
|
|
del sys.modules["snob_lib"]
|
|
else:
|
|
sys.modules["snob_lib"] = previous_snob
|
|
|
|
self.assertEqual([["insights/api/feature_flags.py"]], seen_changed_files)
|
|
self.assertEqual(
|
|
{"status": "ok", "tests": ["insights/api/test/test_feature_flags.py"], "count": 1},
|
|
result,
|
|
)
|
|
|
|
def test_signal_handler_change_expands_to_app_and_api_tests(self) -> None:
|
|
with tempfile.TemporaryDirectory() as root:
|
|
tmp_path = Path(root)
|
|
selection = _load_selection_module()
|
|
selection.REPO_ROOT = tmp_path
|
|
|
|
# A signal handler file
|
|
handler = tmp_path / "insights" / "models" / "signal_handlers.py"
|
|
handler.parent.mkdir(parents=True)
|
|
handler.write_text("from django.db.models.signals import post_save\npost_save.connect(my_handler)\n")
|
|
|
|
# A test in the same app
|
|
test_path = tmp_path / "insights" / "models" / "test" / "test_models.py"
|
|
test_path.parent.mkdir(parents=True)
|
|
test_path.write_text("def test_model(): pass\n")
|
|
|
|
# An API test elsewhere
|
|
api_test = tmp_path / "insights" / "api" / "test" / "test_something.py"
|
|
api_test.parent.mkdir(parents=True)
|
|
api_test.write_text("from rest_framework.test import APIClient\nclient = APIClient()\n")
|
|
|
|
features_by_path = selection.classify_tests()
|
|
result = selection.ast_select_tests(
|
|
["insights/models/signal_handlers.py"],
|
|
features_by_path,
|
|
)
|
|
|
|
self.assertIn("signal_handler_app:insights/models", result.groups)
|
|
self.assertIn("insights/models/test/test_models.py", result.tests)
|
|
|
|
def test_middleware_change_expands_to_api_tests(self) -> None:
|
|
selection = _load_selection_module()
|
|
|
|
result = selection.ast_select_tests(
|
|
["insights/gzip_middleware.py"],
|
|
{
|
|
"insights/api/test/test_capture.py": selection.TestFeatures(
|
|
path="insights/api/test/test_capture.py",
|
|
imports_api_client=True,
|
|
),
|
|
"insights/models/test/test_utils.py": selection.TestFeatures(
|
|
path="insights/models/test/test_utils.py",
|
|
),
|
|
},
|
|
)
|
|
|
|
self.assertIn("middleware_api_tests", result.groups)
|
|
self.assertIn("insights/api/test/test_capture.py", result.tests)
|
|
# Non-API tests are NOT included by middleware expansion
|
|
self.assertNotIn("insights/models/test/test_utils.py", result.groups.get("middleware_api_tests", []))
|
|
|
|
def test_db_router_change_expands_to_api_tests(self) -> None:
|
|
selection = _load_selection_module()
|
|
|
|
result = selection.ast_select_tests(
|
|
["insights/product_db_router.py"],
|
|
{
|
|
"insights/api/test/test_user.py": selection.TestFeatures(
|
|
path="insights/api/test/test_user.py",
|
|
imports_api_client=True,
|
|
),
|
|
},
|
|
)
|
|
|
|
self.assertIn("db_router_api_tests", result.groups)
|
|
self.assertIn("insights/api/test/test_user.py", result.tests)
|
|
|
|
def test_same_app_fallback_includes_sibling_tests(self) -> None:
|
|
selection = _load_selection_module()
|
|
|
|
result = selection.ast_select_tests(
|
|
["products/surveys/backend/models.py"],
|
|
{
|
|
"products/surveys/backend/test/test_api.py": selection.TestFeatures(
|
|
path="products/surveys/backend/test/test_api.py",
|
|
),
|
|
"products/experiments/backend/test/test_api.py": selection.TestFeatures(
|
|
path="products/experiments/backend/test/test_api.py",
|
|
),
|
|
},
|
|
)
|
|
|
|
# Same-app tests included
|
|
self.assertIn("products/surveys/backend/test/test_api.py", result.tests)
|
|
# Different app tests NOT included
|
|
self.assertNotIn("products/experiments/backend/test/test_api.py", result.tests)
|
|
|
|
def test_too_many_files_signals_full_run(self) -> None:
|
|
selection = _load_selection_module()
|
|
|
|
many_files = [f"insights/models/model_{i}.py" for i in range(60)]
|
|
result = selection.ast_select_tests(many_files, {})
|
|
|
|
self.assertTrue(any("too many changed files" in r for r in result.full_run_reasons))
|
|
|
|
def test_high_fanout_file_signals_full_run(self) -> None:
|
|
selection = _load_selection_module()
|
|
|
|
original_path = selection.HIGH_FANOUT_PATH
|
|
try:
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f:
|
|
f.write("insights/redis.py\ninsights/models/team/team.py\n")
|
|
selection.HIGH_FANOUT_PATH = Path(f.name)
|
|
|
|
result = selection.ast_select_tests(["insights/redis.py"], {})
|
|
self.assertTrue(any("high-fanout" in r for r in result.full_run_reasons))
|
|
finally:
|
|
selection.HIGH_FANOUT_PATH = original_path
|
|
|
|
def test_changed_tests_do_not_trigger_full_run_patterns(self) -> None:
|
|
selection = _load_selection_module()
|
|
|
|
result = selection.ast_select_tests(
|
|
["insights/test/test_version_requirement.py"],
|
|
{
|
|
"insights/test/test_version_requirement.py": selection.TestFeatures(
|
|
path="insights/test/test_version_requirement.py"
|
|
)
|
|
},
|
|
)
|
|
|
|
self.assertEqual([], result.full_run_reasons)
|
|
self.assertEqual({"changed_tests": ["insights/test/test_version_requirement.py"]}, result.groups)
|
|
|
|
def test_segments_for_test_file_mirrors_matrix_partition(self) -> None:
|
|
selection = _load_selection_module()
|
|
|
|
self.assertEqual(selection.segments_for_test_file("insights/models/test_a.py"), frozenset({"core"}))
|
|
# POE patterns run in both the Core matrix and the person-on-events matrix.
|
|
self.assertEqual(selection.segments_for_test_file("insights/datastore/test_b.py"), frozenset({"core", "poe"}))
|
|
self.assertEqual(selection.segments_for_test_file("insights/temporal/tests/test_c.py"), frozenset({"temporal"}))
|
|
# Explicitly ignored by the Core invocation, and not a draft-narrowable matrix.
|
|
self.assertEqual(selection.segments_for_test_file("insights/dags/test_e.py"), frozenset())
|
|
# Product/turbo tests are not part of any draft-narrowable Django matrix.
|
|
self.assertEqual(selection.segments_for_test_file("products/warehouse_sources/backend/test_d.py"), frozenset())
|
|
|
|
def test_narrowable_baseline_excludes_turbo_product_tests(self) -> None:
|
|
selection = _load_selection_module()
|
|
|
|
durations = {
|
|
"insights/models/test_a.py::t1": 100.0,
|
|
"insights/datastore/test_b.py::t2": 70.0,
|
|
"insights/temporal/tests/test_c.py::t3": 210.0,
|
|
"insights/dags/test_e.py::t5": 42.0, # ignored path
|
|
"products/warehouse_sources/backend/test_d.py::t4": 100_000.0, # turbo-tests
|
|
}
|
|
|
|
# Only the Core/POE/Temporal universe counts; the huge product test and the ignored
|
|
# path are excluded, so a draft can never be credited with skipping them.
|
|
self.assertEqual(selection.narrowable_baseline_seconds(durations), 380.0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|