test: guard spec-based mock constructors with ast-grep (#43232)

Co-authored-by: Codex <codex@openai.com>

.github/workflows/main-ci.yml

@@ -63,9 +63,11 @@ jobs:
               - 'api/**'
               - 'scripts/ast_grep_guard.py'
               - 'scripts/check_no_new_getattr.py'
+              - 'scripts/check_no_spec_mock.py'
               - 'scripts/check_no_new_controller_sqlalchemy.py'
               - 'scripts/lint_controller_sqlalchemy.py'
               - 'scripts/ast_grep_rules/no_new_getattr.yml'
+              - 'scripts/ast_grep_rules/no_spec_mock.yml'
               - 'scripts/ast_grep_rules/no_new_controller_sqlalchemy.yml'
               - 'scripts/ast_grep_rules/no_new_session_mock.yml'
               - '.github/workflows/style.yml'
@@ -162,10 +164,12 @@ jobs:
               - 'api/**'
               - 'scripts/ast_grep_guard.py'
               - 'scripts/check_no_new_getattr.py'
+              - 'scripts/check_no_spec_mock.py'
               - 'scripts/check_no_new_controller_sqlalchemy.py'
               - 'scripts/check_no_new_session_mock.py'
               - 'scripts/lint_controller_sqlalchemy.py'
               - 'scripts/ast_grep_rules/no_new_getattr.yml'
+              - 'scripts/ast_grep_rules/no_spec_mock.yml'
               - 'scripts/ast_grep_rules/no_new_controller_sqlalchemy.yml'
               - 'scripts/ast_grep_rules/no_new_session_mock.yml'
               - '.github/workflows/style.yml'

.github/workflows/style.yml

@@ -93,6 +93,10 @@ jobs:
         if: inputs.run-python-style
         run: uv run --project api python scripts/check_no_new_session_mock.py --base-rev "${{ inputs.base-rev }}"
 
+      - name: Run No Spec Mock Guard
+        if: inputs.run-python-style
+        run: uv run --project api python scripts/check_no_spec_mock.py
+
       - name: Run Type Checks
         if: inputs.run-python-style
         env:

Makefile

@@ -80,6 +80,7 @@ lint:
 	@uv run --project api --dev ruff format ./api
 	@uv run --project api --dev ruff check --fix ./api
 	@$(MAKE) api-contract-lint
+	@uv run --project api python scripts/check_no_spec_mock.py
 	@uv run --directory api --dev lint-imports
 	@uv run --project api --dev dotenv-linter ./api/.env.example ./web/.env.example
 	@echo "✅ Linting complete"

api/tests/test_containers_integration_tests/services/workspace/test_invitations.py

@@ -2,7 +2,7 @@
 
 import json
 from hashlib import sha256
-from unittest.mock import Mock
+from unittest.mock import create_autospec
 from uuid import uuid4
 
 import pytest
@@ -50,7 +50,7 @@ def test_invitation_token_round_trip_and_acceptance(
     tenant, owner, invited = workspace
     invited.status = status
     db_session_with_containers.commit()
-    send = Mock(spec=gateways.send_invite_member_mail_task.delay)
+    send = create_autospec(gateways.send_invite_member_mail_task.delay, instance=True)
     monkeypatch.setattr(gateways.send_invite_member_mail_task, "delay", send)
     services = application_services()
 

api/tests/unit_tests/account_domain.py

@@ -2,7 +2,7 @@
 
 from dataclasses import dataclass
 from datetime import datetime
-from unittest.mock import Mock
+from unittest.mock import Mock, create_autospec
 
 from sqlalchemy.orm import Session, sessionmaker
 
@@ -39,14 +39,14 @@ class AccountDomain:
 def build_account_domain(session_factory: sessionmaker[Session]) -> AccountDomain:
     repository = SQLAlchemyAccountRepository(session_factory)
     workspaces = WorkspaceRepository(session_factory)
-    policy = Mock(spec=AccountLifecyclePolicy)
+    policy = create_autospec(AccountLifecyclePolicy, instance=True)
     policy.is_registration_allowed.return_value = True
     policy.has_account_capacity.return_value = True
     policy.is_workspace_creation_allowed.return_value = True
     policy.has_workspace_capacity.return_value = True
     policy.get_email_freeze_type.return_value = None
     policy.validate_timezone.side_effect = lambda timezone: timezone
-    access = Mock(spec=WorkspaceMemberAccessGateway)
+    access = create_autospec(WorkspaceMemberAccessGateway, instance=True)
     access.rbac_enabled = False
     access.permission_keys.return_value = {"workspace.member.manage", "workspace.role.manage"}
     access.is_owner.return_value = False
@@ -59,8 +59,8 @@ def build_account_domain(session_factory: sessionmaker[Session]) -> AccountDomai
         memberships=workspaces,
         members=members,
     )
-    sessions = Mock(spec=AccountSessionGateway)
-    security = Mock(spec=ConsoleAuthSecurityGateway)
+    sessions = create_autospec(AccountSessionGateway, instance=True)
+    security = create_autospec(ConsoleAuthSecurityGateway, instance=True)
     accounts = AccountService(
         accounts=repository,
         policies=policy,
@@ -70,7 +70,7 @@ def build_account_domain(session_factory: sessionmaker[Session]) -> AccountDomai
         security=security,
         now=lambda: datetime(2026, 1, 1),
     )
-    delivery = Mock(spec=WorkspaceInvitationDelivery)
+    delivery = create_autospec(WorkspaceInvitationDelivery, instance=True)
     delivery.requires_capacity_check = False
     delivery.create.return_value = "invitation-token"
     invitations = WorkspaceInvitationService(

api/tests/unit_tests/commands/test_check_no_spec_mock.py

@@ -0,0 +1,85 @@
+"""Exercise the real ast-grep gate, including constructor aliases and false positives."""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+REPO_ROOT = Path(__file__).resolve().parents[4]
+SCRIPT = REPO_ROOT / "scripts" / "check_no_spec_mock.py"
+
+
+def scan(tmp_path: Path, source: str) -> subprocess.CompletedProcess[str]:
+    path = tmp_path / "sample.py"
+    path.write_text(source)
+    return subprocess.run([sys.executable, str(SCRIPT), str(path)], text=True, capture_output=True, check=False)
+
+
+@pytest.mark.parametrize(
+    "constructor",
+    ["Mock", "MagicMock", "AsyncMock", "NonCallableMock", "NonCallableMagicMock", "mock.Mock", "mocker.MagicMock"],
+)
+@pytest.mark.parametrize("arguments", ["spec=Model", "name='value', spec_set=Model", "Model", "Model, name='value'"])
+def test_rejects_spec_constructors(tmp_path: Path, constructor: str, arguments: str) -> None:
+    result = scan(tmp_path, f"value = {constructor}({arguments})\n")
+    assert result.returncode == 1
+    assert "sample.py:1:" in result.stdout + result.stderr
+    assert "no-spec-mock" in result.stdout + result.stderr
+
+
+@pytest.mark.parametrize(
+    "source",
+    [
+        "from unittest.mock import Mock as Fake\nvalue = Fake(spec=Model)\n",
+        "from unittest.mock import (MagicMock as Fake, patch)\nvalue = Fake(Model)\n",
+        "factory = mock.AsyncMock\nvalue = factory(spec_set=Client)\n",
+        "from unittest import mock as doubles\nvalue = doubles.NonCallableMock(Model)\n",
+        "value = Mock(**{'spec': Model})\n",
+        'value = Mock(**{"name": "fixture", "spec_set": Model})\n',
+        "value = unittest.mock.MagicMock(\n    name='value',\n    spec=Model,\n)\n",
+        "value = Mock(\n    # A positional argument is also a spec.\n    Model,\n)\n",
+    ],
+)
+def test_rejects_aliases_and_multiline_calls(tmp_path: Path, source: str) -> None:
+    result = scan(tmp_path, source)
+    assert result.returncode == 1
+    assert "no-spec-mock" in result.stdout + result.stderr
+
+
+@pytest.mark.parametrize(
+    "source",
+    [
+        "value = Mock()\n",
+        "value = Mock(name='value', return_value=Model())\n",
+        "value = MagicMock(side_effect=RuntimeError('failed'))\n",
+        "value = Mock(return_value=factory(spec=Model))\n",
+        "value = Mock(\n    # A method return value is allowed.\n    return_value=Model(),\n)\n",
+        "value = Mock(**kwargs)\n",
+        "value = MagicMock(**{'return_value.run.return_value': iter([])})\n",
+        "from unittest.mock import Mock as Fake\nvalue = Fake(return_value=3)\n",
+        "value = Mock(**{'return_value': {'spec': Model}})\n",
+        "value = create_autospec(Port, instance=True, spec_set=True)\n",
+        "value = Schema(spec=Model)\n",
+        "from other_library import Mock as Builder\nvalue = Builder(spec=Model)\n",
+        "# value = Mock(spec=Model)\ntext = 'MagicMock(Model)'\n",
+    ],
+)
+def test_allows_non_spec_mocks_and_unrelated_syntax(tmp_path: Path, source: str) -> None:
+    result = scan(tmp_path, source)
+    assert result.returncode == 0, result.stdout + result.stderr
+
+
+def test_checks_existing_calls_without_a_diff(tmp_path: Path) -> None:
+    result = scan(tmp_path, "def existing_fixture():\n    return Mock(\n        spec=Model,\n    )\n")
+    assert result.returncode == 1
+    assert "sample.py:2:" in result.stdout + result.stderr
+
+
+def test_missing_scan_path_fails(tmp_path: Path) -> None:
+    result = subprocess.run(
+        [sys.executable, str(SCRIPT), str(tmp_path / "missing")], text=True, capture_output=True, check=False
+    )
+    assert result.returncode != 0

api/tests/unit_tests/conftest.py

@@ -6,7 +6,7 @@ from collections.abc import Callable, Iterator
 from pathlib import Path
 from tempfile import TemporaryDirectory
 from typing import TYPE_CHECKING
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock, create_autospec, patch
 
 import pytest
 from flask import Flask
@@ -315,7 +315,6 @@ def account_application_services(
     sqlite_session_factory: sessionmaker[Session], account_domain: AccountDomain
 ) -> ApplicationServices:
     from dataclasses import replace
-    from unittest.mock import Mock
 
     from enums import DeploymentEdition
     from extensions.ext_application_services import build_application_services
@@ -325,7 +324,7 @@ def account_application_services(
         database_client=sqlite_session_factory,
         deployment_edition=DeploymentEdition.COMMUNITY,
         initialization_password="",
-        redis=Mock(spec=RedisClientWrapper),
+        redis=create_autospec(RedisClientWrapper, instance=True),
     )
     return replace(
         services,

api/tests/unit_tests/controllers/console/workspace/test_members.py

@@ -3,7 +3,7 @@ from datetime import datetime
 from http import HTTPStatus
 from inspect import unwrap
 from typing import NamedTuple, override
-from unittest.mock import Mock, patch
+from unittest.mock import Mock, create_autospec, patch
 
 import pytest
 from flask import Flask
@@ -145,13 +145,13 @@ def services(monkeypatch: pytest.MonkeyPatch) -> Mock:
     from extensions.ext_application_services import AccountServices, ApplicationServices
     from services.workspace.member_service import WorkspaceInvitationService, WorkspaceOwnerTransferService
 
-    services = Mock(spec=ApplicationServices)
-    services.accounts = Mock(spec=AccountServices)
-    services.workspaces = Mock(spec=WorkspaceServices)
-    services.workspaces.invitations = Mock(spec=WorkspaceInvitationService)
-    services.workspaces.owner_transfer = Mock(spec=WorkspaceOwnerTransferService)
-    services.workspaces.members = Mock(spec=WorkspaceMemberService)
-    services.workspaces.member_queries = Mock(spec=WorkspaceMemberQueryService)
+    services = create_autospec(ApplicationServices, instance=True)
+    services.accounts = create_autospec(AccountServices, instance=True)
+    services.workspaces = create_autospec(WorkspaceServices, instance=True)
+    services.workspaces.invitations = create_autospec(WorkspaceInvitationService, instance=True)
+    services.workspaces.owner_transfer = create_autospec(WorkspaceOwnerTransferService, instance=True)
+    services.workspaces.members = create_autospec(WorkspaceMemberService, instance=True)
+    services.workspaces.member_queries = create_autospec(WorkspaceMemberQueryService, instance=True)
     import importlib
 
     monkeypatch.setattr(

api/tests/unit_tests/controllers/console/workspace/test_workspace.py

@@ -6,7 +6,7 @@ from datetime import datetime
 from http import HTTPStatus
 from inspect import unwrap
 from io import BytesIO
-from unittest.mock import Mock
+from unittest.mock import Mock, create_autospec
 
 import pytest
 from flask import Flask
@@ -58,9 +58,9 @@ class Services:
 @pytest.fixture
 def workspaces(monkeypatch: pytest.MonkeyPatch) -> WorkspaceMocks:
     services = WorkspaceMocks(
-        management=Mock(spec=WorkspaceService),
-        queries=Mock(spec=WorkspaceQueryService),
-        provisioning=Mock(spec=WorkspaceProvisioningService),
+        management=create_autospec(WorkspaceService, instance=True),
+        queries=create_autospec(WorkspaceQueryService, instance=True),
+        provisioning=create_autospec(WorkspaceProvisioningService, instance=True),
     )
     monkeypatch.setattr(controller, "application_services", lambda: Services(services))
     return services

api/tests/unit_tests/controllers/inner_api/workspace/test_workspace.py

@@ -4,7 +4,7 @@ from collections.abc import Callable
 from dataclasses import dataclass
 from datetime import datetime
 from inspect import unwrap
-from unittest.mock import Mock
+from unittest.mock import Mock, create_autospec
 
 import pytest
 from flask import Flask
@@ -35,10 +35,10 @@ class Services:
 
 @pytest.fixture
 def provisioning(monkeypatch: pytest.MonkeyPatch) -> Mock:
-    provisioning = Mock(spec=WorkspaceProvisioningService)
+    provisioning = create_autospec(WorkspaceProvisioningService, instance=True)
     services = WorkspaceServices(
-        queries=Mock(spec=WorkspaceQueryService),
-        management=Mock(spec=WorkspaceService),
+        queries=create_autospec(WorkspaceQueryService, instance=True),
+        management=create_autospec(WorkspaceService, instance=True),
         provisioning=provisioning,
         members=Mock(),
         member_queries=Mock(),

api/tests/unit_tests/controllers/service_api/dataset/test_metadata.py

@@ -19,7 +19,7 @@ Decorator strategy:
 
 import uuid
 from inspect import unwrap
-from unittest.mock import ANY, Mock, patch
+from unittest.mock import ANY, Mock, create_autospec, patch
 
 import pytest
 from flask import Flask, request
@@ -69,14 +69,16 @@ def test_metadata_permission_error_http_contract(
     monkeypatch.setattr(
         metadata_module.DatasetService,
         "get_dataset_for_tenant",
-        Mock(spec=metadata_module.DatasetService.get_dataset_for_tenant, return_value=mock_dataset),
+        create_autospec(
+            metadata_module.DatasetService.get_dataset_for_tenant, return_value=mock_dataset, instance=True
+        ),
     )
     monkeypatch.setattr(
         metadata_module.DatasetService,
         "check_dataset_permission",
-        Mock(spec=metadata_module.DatasetService.check_dataset_permission, side_effect=error),
+        create_autospec(metadata_module.DatasetService.check_dataset_permission, side_effect=error, instance=True),
     )
-    metadata = Mock(spec=MetadataService)
+    metadata = create_autospec(MetadataService, instance=True)
     services = Mock()
     services.knowledge.metadata = metadata
     monkeypatch.setattr(metadata_module, "application_services", lambda: services)

api/tests/unit_tests/extensions/test_ext_application_services.py

@@ -7,7 +7,7 @@ from decimal import Decimal
 from io import BytesIO
 from types import SimpleNamespace
 from typing import cast
-from unittest.mock import MagicMock, call, patch
+from unittest.mock import MagicMock, call, create_autospec, patch
 from uuid import uuid4
 
 import httpx
@@ -494,7 +494,7 @@ def test_build_application_services_wires_web_authentication_boundary(
         database_client=sqlite_session_factory,
         deployment_edition=DeploymentEdition.ENTERPRISE,
         initialization_password="",
-        redis=MagicMock(spec=RedisClientWrapper),
+        redis=create_autospec(RedisClientWrapper, instance=True),
     )
 
     assert isinstance(services.web_authentication, WebAuthenticationService)
@@ -520,7 +520,7 @@ def test_build_application_services_reuses_installed_app_dependencies(
         database_client=sqlite_session_factory,
         deployment_edition=DeploymentEdition.COMMUNITY,
         initialization_password="",
-        redis=MagicMock(spec=RedisClientWrapper),
+        redis=create_autospec(RedisClientWrapper, instance=True),
     )
 
     assert isinstance(services.installed_apps.access, InstalledAppAccessService)

api/tests/unit_tests/repositories/workspace/test_workspace_repository.py

@@ -2,7 +2,7 @@
 
 from collections.abc import Sequence
 from datetime import datetime, timedelta
-from unittest.mock import Mock
+from unittest.mock import Mock, create_autospec
 
 import pytest
 from sqlalchemy import Engine, event, select
@@ -152,7 +152,7 @@ def test_sessions_are_closed_before_feature_and_plan_io(
 
     event.listen(sqlite_engine, "checkout", checkout)
     event.listen(sqlite_engine, "checkin", checkin)
-    features = Mock(spec=WorkspaceFeatureGateway)
+    features = create_autospec(WorkspaceFeatureGateway, instance=True)
 
     def get_features(workspace_id: str) -> WorkspaceFeatures:
         assert not checked_out
@@ -163,7 +163,9 @@ def test_sessions_are_closed_before_feature_and_plan_io(
         return WorkspaceFeatures(False, EffectiveCreditPool())
 
     features.get_features.side_effect = get_features
-    service = WorkspaceService(workspaces=seeded, features=features, logos=Mock(spec=WorkspaceLogoGateway))
+    service = WorkspaceService(
+        workspaces=seeded, features=features, logos=create_autospec(WorkspaceLogoGateway, instance=True)
+    )
     context = RequestContext("request", None, "a1", "w1")
     try:
         service.rename(context, "Committed")

api/tests/unit_tests/services/account/test_email_registration_adapters.py

@@ -1,5 +1,5 @@
 from collections.abc import Callable
-from unittest.mock import MagicMock, Mock, call, patch
+from unittest.mock import MagicMock, call, create_autospec, patch
 
 import pytest
 from redis import RedisError
@@ -73,7 +73,7 @@ def test_security_gateway_preserves_shared_ip_limit(
     commands.side_effect = [None, str(count), None, True] if limited else [None, str(count), True, True]
     gateway = RedisEmailRegistrationSecurityGateway(
         redis=redis,
-        login_security=Mock(spec=ConsoleAuthSecurityGateway),
+        login_security=create_autospec(ConsoleAuthSecurityGateway, instance=True),
         verification_failure_limit=5,
         verification_lockout_duration=600,
     )
@@ -128,11 +128,11 @@ def test_security_gateway_uses_registration_and_login_keys(
 
 @pytest.mark.parametrize(("count", "limited"), [(5, False), (6, True)])
 def test_registration_verification_limit_preserves_threshold(count: int, limited: bool) -> None:
-    redis = Mock(spec=RedisClientWrapper)
+    redis = create_autospec(RedisClientWrapper, instance=True)
     redis.get.return_value = str(count)
     gateway = RedisEmailRegistrationSecurityGateway(
         redis=redis,
-        login_security=Mock(spec=ConsoleAuthSecurityGateway),
+        login_security=create_autospec(ConsoleAuthSecurityGateway, instance=True),
         verification_failure_limit=5,
         verification_lockout_duration=600,
     )
@@ -141,7 +141,7 @@ def test_registration_verification_limit_preserves_threshold(count: int, limited
 
 
 def test_registration_security_preserves_behavior_when_redis_is_unavailable() -> None:
-    redis = Mock(spec=RedisClientWrapper)
+    redis = create_autospec(RedisClientWrapper, instance=True)
     redis.get.side_effect = RedisError("offline")
     redis.delete.side_effect = RedisError("offline")
     gateway = RedisEmailRegistrationSecurityGateway(
@@ -182,7 +182,7 @@ def test_registration_gateway_preserves_domain_errors_and_translates_seat_limit(
     service_error: Exception,
     application_error: type[Exception],
 ) -> None:
-    gateway = AccountLifecycleRegistrationGateway(accounts=Mock(spec=AccountService))
+    gateway = AccountLifecycleRegistrationGateway(accounts=create_autospec(AccountService, instance=True))
 
     with patch.object(gateway._accounts, "create_account_and_tenant", side_effect=service_error):
         with pytest.raises(application_error) as raised:

api/tests/unit_tests/services/account/test_identity.py

@@ -1,5 +1,5 @@
 from datetime import datetime, timedelta
-from unittest.mock import Mock
+from unittest.mock import create_autospec
 
 import pytest
 from redis import RedisError
@@ -66,7 +66,7 @@ def test_activity_refresh_is_gated_but_fails_open(
     )
     sqlite_session.commit()
     account_id, old_active = account.id, account.last_active_at
-    redis = Mock(spec=RedisClientWrapper)
+    redis = create_autospec(RedisClientWrapper, instance=True)
     if isinstance(redis_claim, RedisError):
         redis.set.side_effect = redis_claim
     else:
@@ -83,7 +83,9 @@ def test_identity_rejects_banned_and_missing_accounts(account_domain: AccountDom
     account = Account(name="Banned", email="b@example.com", status=AccountStatus.BANNED)
     sqlite_session.add(account)
     sqlite_session.commit()
-    gateway = AccountIdentityGateway(accounts=account_domain.repository, redis=Mock(spec=RedisClientWrapper))
+    gateway = AccountIdentityGateway(
+        accounts=account_domain.repository, redis=create_autospec(RedisClientWrapper, instance=True)
+    )
     with pytest.raises(Unauthorized, match="banned"):
         gateway.load_user(account.id)
     assert gateway.load_user("missing") is None

api/tests/unit_tests/services/test_account_security_gateway.py

@@ -2,7 +2,7 @@
 
 import logging
 from typing import cast
-from unittest.mock import Mock, call
+from unittest.mock import call, create_autospec
 
 import pytest
 from redis import RedisError
@@ -50,7 +50,7 @@ def test_redis_failures_remain_fail_open_but_are_logged(
 
 @pytest.mark.parametrize("already_frozen", [True, False], ids=["frozen-ip", "lost-hour-counter-race"])
 def test_frozen_ip_and_concurrent_hour_claim_remain_limited(already_frozen: bool) -> None:
-    redis = Mock(spec=RedisClientWrapper)
+    redis = create_autospec(RedisClientWrapper, instance=True)
     values = (
         {"email_send_ip_limit_freeze:127.0.0.1": b"1"}
         if already_frozen

api/tests/unit_tests/services/workspace/test_gateways.py

@@ -1,7 +1,7 @@
 """Workspace policies and notifications with only external clients replaced."""
 
 from collections.abc import Callable
-from unittest.mock import Mock
+from unittest.mock import Mock, create_autospec
 
 import pytest
 
@@ -18,7 +18,7 @@ from tests.unit_tests.account_domain import AccountDomain
 
 @pytest.fixture
 def invitations() -> gateways.WorkspaceInvitationGateway:
-    redis = Mock(spec=RedisClientWrapper)
+    redis = create_autospec(RedisClientWrapper, instance=True)
     return gateways.WorkspaceInvitationGateway(tokens=RedisInvitationTokenStore(redis=redis), redis=redis)
 
 
@@ -56,7 +56,7 @@ def test_cloud_invitation_member_limit(
     config_overrides(DEPLOYMENT_EDITION=DeploymentEdition.CLOUD)
     # Use the repository's current count, not a potentially stale billing size.
     features = Mock(return_value=FeatureModel(members=LimitationModel(size=99, limit=limit)))
-    license_query = Mock(spec=gateways.SystemFeatureService.get_license)
+    license_query = create_autospec(gateways.SystemFeatureService.get_license, instance=True)
     monkeypatch.setattr(gateways.FeatureService, "get_features", features)
     monkeypatch.setattr(gateways.SystemFeatureService, "get_license", license_query)
 
@@ -146,8 +146,8 @@ def test_community_invitation_does_not_query_quotas(
     config_overrides: Callable[..., None],
 ) -> None:
     config_overrides(DEPLOYMENT_EDITION=DeploymentEdition.COMMUNITY)
-    features = Mock(spec=gateways.FeatureService.get_features)
-    license_query = Mock(spec=gateways.SystemFeatureService.get_license)
+    features = create_autospec(gateways.FeatureService.get_features, instance=True)
+    license_query = create_autospec(gateways.SystemFeatureService.get_license, instance=True)
     monkeypatch.setattr(gateways.FeatureService, "get_features", features)
     monkeypatch.setattr(gateways.SystemFeatureService, "get_license", license_query)
 
@@ -212,12 +212,12 @@ def test_single_invitation_quota_preserves_plan_and_license_limits(
 def test_owner_transfer_notifies_both_owners(account_domain: AccountDomain, monkeypatch: pytest.MonkeyPatch) -> None:
     old_owner = account_domain.accounts.create_account("old@example.com", "Old", "en-US")
     new_owner = account_domain.accounts.create_account("new@example.com", "New", "en-US")
-    old_notification = Mock(spec=gateways.send_old_owner_transfer_notify_email_task.delay)
-    new_notification = Mock(spec=gateways.send_new_owner_transfer_notify_email_task.delay)
+    old_notification = create_autospec(gateways.send_old_owner_transfer_notify_email_task.delay, instance=True)
+    new_notification = create_autospec(gateways.send_new_owner_transfer_notify_email_task.delay, instance=True)
     monkeypatch.setattr(gateways.send_old_owner_transfer_notify_email_task, "delay", old_notification)
     monkeypatch.setattr(gateways.send_new_owner_transfer_notify_email_task, "delay", new_notification)
 
-    gateway = gateways.WorkspaceOwnerTransferGateway(redis=Mock(spec=RedisClientWrapper))
+    gateway = gateways.WorkspaceOwnerTransferGateway(redis=create_autospec(RedisClientWrapper, instance=True))
     gateway.notify(old_owner=old_owner, new_owner=new_owner, workspace_name="Workspace")
 
     new_notification.assert_called_once_with(language="en-US", to=new_owner.email, workspace="Workspace")

api/tests/unit_tests/services/workspace/test_member_service.py

@@ -2,7 +2,7 @@
 
 from collections.abc import Callable
 from contextlib import nullcontext
-from unittest.mock import Mock
+from unittest.mock import Mock, create_autospec
 
 import pytest
 from sqlalchemy import event, select
@@ -352,7 +352,7 @@ def test_bulk_invitation_counts_new_accounts_and_members_before_writing(
         "get_license",
         Mock(return_value=LicenseModel(seats=LicenseLimitationModel(enabled=True, size=3, limit=3))),
     )
-    redis = Mock(spec=RedisClientWrapper)
+    redis = create_autospec(RedisClientWrapper, instance=True)
     gateway = WorkspaceInvitationGateway(tokens=RedisInvitationTokenStore(redis=redis), redis=redis)
     workspace_id, owner_id, _ = members
     account_domain.accounts.create_account("outside@example.com", "Outside", "en-US")
@@ -401,7 +401,7 @@ def test_bulk_invitation_retains_partial_results_under_workspace_lock(
 
 @pytest.fixture
 def owner_transfer(account_domain: AccountDomain) -> tuple[WorkspaceOwnerTransferService, Mock]:
-    challenges = Mock(spec=OwnerTransferGateway)
+    challenges = create_autospec(OwnerTransferGateway, instance=True)
     challenges.is_ip_limited.return_value = False
     challenges.is_verification_limited.return_value = False
     challenges.read_token.return_value = OwnerTransferToken("owner@example.com", "123456")
@@ -551,7 +551,7 @@ def rbac_access(
     monkeypatch.setattr(RBACService.Roles, "list", Mock(return_value=Paginated[RBACRole](data=roles)))
     monkeypatch.setattr(RBACService.MemberRoles, "get", get_roles)
     monkeypatch.setattr(gateway, "permission_keys", lambda _workspace_id, _actor_id: {"workspace.role.manage"})
-    replace = Mock(spec=RBACService.MemberRoles.replace)
+    replace = create_autospec(RBACService.MemberRoles.replace, instance=True)
     monkeypatch.setattr(RBACService.MemberRoles, "replace", replace)
     return gateway, replace
 
@@ -590,7 +590,7 @@ def test_rbac_invitation_assigns_requested_role(
         sqlite_session.commit()
     access, replace = rbac_access
     monkeypatch.setattr(access, "permission_keys", lambda _workspace, _actor: {"workspace.member.manage"})
-    sync = Mock(spec=sync_joined_workspace_member_rbac_access_task.delay)
+    sync = create_autospec(sync_joined_workspace_member_rbac_access_task.delay, instance=True)
     monkeypatch.setattr(sync_joined_workspace_member_rbac_access_task, "delay", sync)
     member_service = WorkspaceMemberService(
         workspaces=account_domain.workspaces, accounts=account_domain.repository, access=access

api/tests/unit_tests/services/workspace/test_workspace_file_gateway.py

@@ -1,6 +1,6 @@
 """File adapter keeps admitted identity explicit across the FileService boundary."""
 
-from unittest.mock import Mock
+from unittest.mock import Mock, create_autospec
 
 from machinery.context import RequestContext
 from models.account import Account
@@ -9,7 +9,7 @@ from services.workspace.gateways import WorkspaceFileGateway
 
 
 def test_upload_uses_admitted_account_and_explicit_workspace() -> None:
-    files = Mock(spec=FileService)
+    files = create_autospec(FileService, instance=True)
     files.upload_file.return_value = Mock(id="file-id")
     gateway = WorkspaceFileGateway(files=files)
     context = RequestContext("request", None, "account", "workspace")

api/tests/unit_tests/services/workspace/test_workspace_provisioning.py

@@ -3,7 +3,7 @@
 from collections.abc import Callable
 from datetime import datetime
 from typing import cast
-from unittest.mock import Mock
+from unittest.mock import Mock, create_autospec
 
 import pytest
 from blinker import Signal
@@ -64,7 +64,7 @@ def test_owner_lookup_is_exact_and_tolerates_duplicate_emails(sqlite_session_fac
         session.commit()
         owner_ids = {first.id, second.id}
     repository = SQLAlchemyAccountRepository(sqlite_session_factory)
-    gateway = Mock(spec=WorkspaceProvisioningStore)
+    gateway = create_autospec(WorkspaceProvisioningStore, instance=True)
     service = _service(owners=repository, provisioning=gateway)
     service.create(name="Test", owner_email="owner@example.com")
     assert gateway.provision.call_args.kwargs["owner_id"] in owner_ids
@@ -75,8 +75,8 @@ def test_owner_lookup_is_exact_and_tolerates_duplicate_emails(sqlite_session_fac
 
 @pytest.mark.parametrize("role", ["owner", "unknown", "ADMIN"])
 def test_invalid_roles_are_rejected_before_writes(role: str) -> None:
-    gateway = Mock(spec=WorkspaceProvisioningStore)
-    service = _service(owners=Mock(spec=WorkspaceOwnerQuery), provisioning=gateway)
+    gateway = create_autospec(WorkspaceProvisioningStore, instance=True)
+    service = _service(owners=create_autospec(WorkspaceOwnerQuery, instance=True), provisioning=gateway)
     with pytest.raises(InvalidWorkspaceMemberRoleError):
         service.join_member(
             workspace_id="w", account_id="a", email="a@example.com", role=role, operator_account_id=None
@@ -85,8 +85,8 @@ def test_invalid_roles_are_rejected_before_writes(role: str) -> None:
 
 
 def test_member_operator_is_forwarded() -> None:
-    gateway = Mock(spec=WorkspaceProvisioningStore)
-    service = _service(owners=Mock(spec=WorkspaceOwnerQuery), provisioning=gateway)
+    gateway = create_autospec(WorkspaceProvisioningStore, instance=True)
+    service = _service(owners=create_autospec(WorkspaceOwnerQuery, instance=True), provisioning=gateway)
     service.join_member(
         workspace_id="w", account_id="a", email="a@example.com", role="editor", operator_account_id="operator"
     )
@@ -100,9 +100,9 @@ def test_member_operator_is_forwarded() -> None:
 
 
 def test_missing_owner_is_rejected_before_creation() -> None:
-    owners = Mock(spec=WorkspaceOwnerQuery)
+    owners = create_autospec(WorkspaceOwnerQuery, instance=True)
     owners.find_first_by_email.return_value = None
-    gateway = Mock(spec=WorkspaceProvisioningStore)
+    gateway = create_autospec(WorkspaceProvisioningStore, instance=True)
     service = _service(owners=owners, provisioning=gateway)
     with pytest.raises(WorkspaceOwnerNotFoundError):
         service.create(name="Test", owner_email="owner@example.com")
@@ -111,9 +111,9 @@ def test_missing_owner_is_rejected_before_creation() -> None:
 
 @pytest.mark.parametrize("owner_email", [None, "owner@example.com"])
 def test_create_delegation(owner_email: str | None) -> None:
-    owners = Mock(spec=WorkspaceOwnerQuery)
+    owners = create_autospec(WorkspaceOwnerQuery, instance=True)
     owners.find_first_by_email.return_value = OWNER
-    gateway = Mock(spec=WorkspaceProvisioningStore)
+    gateway = create_autospec(WorkspaceProvisioningStore, instance=True)
     result = CreatedWorkspace("w", "Test", "sandbox", "normal", None, None, None, {})
     gateway.provision.return_value = result
     service = _service(owners=owners, provisioning=gateway)
@@ -302,7 +302,7 @@ def test_competing_initial_workspace_skips_key_and_owner_binding(
     account_domain: AccountDomain, sqlite_session_factory: sessionmaker[Session], monkeypatch: pytest.MonkeyPatch
 ) -> None:
     account = account_domain.accounts.create_account("owner@example.com", "Owner", "en-US")
-    effects = Mock(spec=WorkspaceProvisioningEffects)
+    effects = create_autospec(WorkspaceProvisioningEffects, instance=True)
     effects.prepare.side_effect = [
         WorkspaceCreation("winner", "Owner's Workspace", "winner-key", 123, 0, {}),
         WorkspaceCreation("skipped", "Owner's Workspace", "orphan-key", 123, 0, {}),
@@ -344,7 +344,7 @@ def test_initial_workspace_locks_before_preparation_and_publishes_after_commit(
 ) -> None:
     account = account_domain.accounts.create_account("owner@example.com", "Owner", "en-US")
     phases: list[str] = []
-    effects = Mock(spec=WorkspaceProvisioningEffects)
+    effects = create_autospec(WorkspaceProvisioningEffects, instance=True)
 
     def record_lock(execution: ORMExecuteState) -> None:
         # SQLite exercises persistence; compile the actual ORM statement for
@@ -388,17 +388,18 @@ def _service(
     provisioning: WorkspaceProvisioningStore,
     effects: WorkspaceProvisioningEffects | None = None,
 ) -> WorkspaceProvisioningService:
-    policy = Mock(spec=WorkspaceCreationPolicy)
+    """Use the supplied effects, or prepare autospecced effects when omitted or None."""
+    policy = create_autospec(WorkspaceCreationPolicy, instance=True)
     policy.has_workspace_capacity.return_value = True
     if effects is None:
-        prepared = Mock(spec=WorkspaceProvisioningEffects)
+        prepared = create_autospec(WorkspaceProvisioningEffects, instance=True)
         prepared.prepare.return_value = WorkspaceCreation("w", "Test", "key", 123, 0, {})
-        effects = prepared
+        effects = cast(WorkspaceProvisioningEffects, prepared)
     return WorkspaceProvisioningService(
         effects=effects,
         owners=owners,
         provisioning=provisioning,
         policies=policy,
-        memberships=Mock(spec=WorkspaceMembershipQuery),
-        members=Mock(spec=WorkspaceMemberWriter),
+        memberships=create_autospec(WorkspaceMembershipQuery, instance=True),
+        members=create_autospec(WorkspaceMemberWriter, instance=True),
     )

api/tests/unit_tests/services/workspace/test_workspace_service.py

@@ -2,7 +2,7 @@
 
 from dataclasses import replace
 from datetime import datetime
-from unittest.mock import Mock
+from unittest.mock import Mock, create_autospec
 
 import pytest
 
@@ -32,9 +32,9 @@ WORKSPACE = WorkspaceSnapshot(
 
 @pytest.fixture
 def dependencies() -> tuple[WorkspaceService, Mock, Mock, Mock]:
-    store = Mock(spec=WorkspaceStore)
-    features = Mock(spec=WorkspaceFeatureGateway)
-    logos = Mock(spec=WorkspaceLogoGateway)
+    store = create_autospec(WorkspaceStore, instance=True)
+    features = create_autospec(WorkspaceFeatureGateway, instance=True)
+    logos = create_autospec(WorkspaceLogoGateway, instance=True)
     store.get_for_account.return_value = WORKSPACE
     features.logo_url.return_value = "https://files/workspaces/workspace/webapp-logo"
     features.get_features.return_value = WorkspaceFeatures(True, EffectiveCreditPool())

scripts/ast_grep_rules/no_spec_mock.yml

@@ -0,0 +1,63 @@
+id: no-spec-mock
+language: python
+severity: error
+message: Use a real instance or a concrete test fake instead of a spec-based mock.
+rule:
+  all:
+    - pattern: $CALLEE($$$ARGS)
+    # Inspect direct arguments only; nested calls may have unrelated spec keywords.
+    - has:
+        field: arguments
+        has:
+          any:
+            - all:
+                - kind: keyword_argument
+                - has:
+                    field: name
+                    regex: ^spec(_set)?$
+            - all:
+                - kind: dictionary_splat
+                - has:
+                    kind: dictionary
+                    has:
+                      kind: pair
+                      has:
+                        field: key
+                        regex: ^['"]spec(_set)?['"]$
+            - not:
+                any:
+                  - kind: keyword_argument
+                  - kind: dictionary_splat
+                  - kind: comment
+                  - regex: '^[(),]$'
+    - any:
+        - has:
+            field: function
+            regex: '(^|\.)(Mock|MagicMock|AsyncMock|NonCallableMock|NonCallableMagicMock)$'
+        - inside:
+            kind: module
+            stopBy: end
+            has:
+              stopBy: end
+              all:
+                - pattern:
+                    context: from unittest.mock import $ORIGINAL as $CALLEE
+                    selector: aliased_import
+                - has:
+                    field: name
+                    regex: '^(Mock|MagicMock|AsyncMock|NonCallableMock|NonCallableMagicMock)$'
+                - inside:
+                    kind: import_from_statement
+                    has:
+                      field: module_name
+                      regex: '^unittest\.mock$'
+        - inside:
+            kind: module
+            stopBy: end
+            has:
+              stopBy: end
+              all:
+                - pattern: $CALLEE = $ORIGINAL
+                - has:
+                    field: right
+                    regex: '(^|\.)(Mock|MagicMock|AsyncMock|NonCallableMock|NonCallableMagicMock)$'

scripts/check_no_spec_mock.py

@@ -0,0 +1,47 @@
+#!/usr/bin/env python3
+"""Reject spec-based mock constructors in the API, including unchanged files.
+
+Use real model/SDK instances or concrete test fakes. The rule covers explicit
+spec/spec_set keywords, positional specs, qualified constructors, import aliases,
+and direct constructor aliases. It does not prohibit ordinary method mocks or
+create_autospec. Optional paths allow a focused local scan; CI scans all of api/.
+"""
+
+from __future__ import annotations
+
+import argparse
+import subprocess
+import sys
+from pathlib import Path
+
+from ast_grep_guard import REPO_ROOT, resolve_ast_grep_command, rule_path
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("paths", nargs="*", type=Path, help="Files or directories to scan (default: api/).")
+    args = parser.parse_args()
+    paths = args.paths or [REPO_ROOT / "api"]
+    try:
+        for path in paths:
+            if not path.exists():
+                raise RuntimeError(f"Scan path does not exist: {path}")
+        return subprocess.run(
+            [
+                *resolve_ast_grep_command(),
+                "scan",
+                "--rule",
+                str(rule_path("no_spec_mock.yml")),
+                "--report-style",
+                "short",
+                *(str(path) for path in paths),
+            ],
+            check=False,
+        ).returncode
+    except (OSError, RuntimeError) as exc:
+        sys.stderr.write(f"{exc}\n")
+        return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())