Add test safeguards to prevent production database pollution (bd-2c5a)
- Add failIfProductionDatabase() check in Go test helpers - Add temp directory verification in RPC test setup - Create conftest.py with pytest safety checks for Python tests - Add BEADS_TEST_MODE env var to mark test execution - Tests now fail fast if they detect production .beads/ usage This prevents test issues from polluting the production database like the incident on Nov 7, 2025 where 29+ test issues were created in .beads/beads.db instead of isolated test databases. Resolves: bd-2c5a Amp-Thread-ID: https://ampcode.com/threads/T-635a8807-1120-4122-a0cb-4c21970362ce Co-authored-by: Amp <amp@ampcode.com>
This commit is contained in:
@@ -13,11 +13,69 @@ import (
|
||||
|
||||
const windowsOS = "windows"
|
||||
|
||||
// ensureTestMode sets BEADS_TEST_MODE environment variable to prevent production pollution
|
||||
func ensureTestMode(t *testing.T) {
|
||||
t.Helper()
|
||||
os.Setenv("BEADS_TEST_MODE", "1")
|
||||
t.Cleanup(func() {
|
||||
os.Unsetenv("BEADS_TEST_MODE")
|
||||
})
|
||||
}
|
||||
|
||||
// failIfProductionDatabase checks if the database path is in a production directory
|
||||
// and fails the test to prevent test pollution (bd-2c5a)
|
||||
func failIfProductionDatabase(t *testing.T, dbPath string) {
|
||||
t.Helper()
|
||||
|
||||
// CRITICAL (bd-2c5a): Set test mode flag
|
||||
ensureTestMode(t)
|
||||
|
||||
// Get absolute path for comparison
|
||||
absPath, err := filepath.Abs(dbPath)
|
||||
if err != nil {
|
||||
t.Logf("Warning: Could not get absolute path for %s: %v", dbPath, err)
|
||||
return
|
||||
}
|
||||
|
||||
// Check if database is in a directory that contains .git
|
||||
dir := filepath.Dir(absPath)
|
||||
for {
|
||||
gitPath := filepath.Join(dir, ".git")
|
||||
if _, err := os.Stat(gitPath); err == nil {
|
||||
// Found .git directory - check if this is a test or production database
|
||||
beadsPath := filepath.Join(dir, ".beads")
|
||||
if strings.HasPrefix(absPath, beadsPath) {
|
||||
// Database is in .beads/ directory of a git repository
|
||||
// This is ONLY allowed if we're in a temp directory
|
||||
if !strings.Contains(absPath, os.TempDir()) {
|
||||
t.Fatalf("PRODUCTION DATABASE POLLUTION DETECTED (bd-2c5a):\n"+
|
||||
" Database: %s\n"+
|
||||
" Git repo: %s\n"+
|
||||
" Tests MUST use t.TempDir() or tempfile to create isolated databases.\n"+
|
||||
" This prevents test issues from polluting the production database.",
|
||||
absPath, dir)
|
||||
}
|
||||
}
|
||||
break
|
||||
}
|
||||
|
||||
parent := filepath.Dir(dir)
|
||||
if parent == dir {
|
||||
// Reached filesystem root
|
||||
break
|
||||
}
|
||||
dir = parent
|
||||
}
|
||||
}
|
||||
|
||||
// newTestStore creates a SQLite store with issue_prefix configured (bd-166)
|
||||
// This prevents "database not initialized" errors in tests
|
||||
func newTestStore(t *testing.T, dbPath string) *sqlite.SQLiteStorage {
|
||||
t.Helper()
|
||||
|
||||
// CRITICAL (bd-2c5a): Ensure we're not polluting production database
|
||||
failIfProductionDatabase(t, dbPath)
|
||||
|
||||
if err := os.MkdirAll(filepath.Dir(dbPath), 0755); err != nil {
|
||||
t.Fatalf("Failed to create database directory: %v", err)
|
||||
}
|
||||
@@ -42,6 +100,9 @@ func newTestStore(t *testing.T, dbPath string) *sqlite.SQLiteStorage {
|
||||
func newTestStoreWithPrefix(t *testing.T, dbPath string, prefix string) *sqlite.SQLiteStorage {
|
||||
t.Helper()
|
||||
|
||||
// CRITICAL (bd-2c5a): Ensure we're not polluting production database
|
||||
failIfProductionDatabase(t, dbPath)
|
||||
|
||||
if err := os.MkdirAll(filepath.Dir(dbPath), 0755); err != nil {
|
||||
t.Fatalf("Failed to create database directory: %v", err)
|
||||
}
|
||||
|
||||
73
integrations/beads-mcp/tests/conftest.py
Normal file
73
integrations/beads-mcp/tests/conftest.py
Normal file
@@ -0,0 +1,73 @@
|
||||
"""Pytest configuration and fixtures for beads-mcp tests.
|
||||
|
||||
This module provides safety checks to prevent test pollution in production databases (bd-2c5a).
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
def pytest_configure(config):
|
||||
"""Called before test collection starts - ensure we're not polluting production."""
|
||||
# CRITICAL (bd-2c5a): Prevent tests from polluting production database
|
||||
|
||||
# Set test mode flag
|
||||
os.environ["BEADS_TEST_MODE"] = "1"
|
||||
|
||||
# Get the project root (where .git exists)
|
||||
current_dir = Path(__file__).parent.absolute()
|
||||
project_root = current_dir
|
||||
|
||||
while project_root.parent != project_root:
|
||||
if (project_root / ".git").exists():
|
||||
break
|
||||
project_root = project_root.parent
|
||||
|
||||
# If BEADS_DB or BEADS_WORKING_DIR point to production .beads/, fail immediately
|
||||
beads_db = os.environ.get("BEADS_DB", "")
|
||||
working_dir = os.environ.get("BEADS_WORKING_DIR", "")
|
||||
|
||||
production_beads = str(project_root / ".beads")
|
||||
|
||||
if beads_db and beads_db.startswith(production_beads):
|
||||
pytest.exit(
|
||||
f"PRODUCTION DATABASE POLLUTION DETECTED (bd-2c5a):\n"
|
||||
f" BEADS_DB={beads_db}\n"
|
||||
f" Production .beads/: {production_beads}\n"
|
||||
f" Tests MUST use isolated temp databases.\n"
|
||||
f" Remove BEADS_DB env var or point it to a temp directory.",
|
||||
returncode=1,
|
||||
)
|
||||
|
||||
if working_dir and working_dir.startswith(str(project_root)):
|
||||
# Working dir in project is OK ONLY if it's not the project root itself
|
||||
if Path(working_dir).resolve() == project_root.resolve():
|
||||
pytest.exit(
|
||||
f"PRODUCTION DATABASE POLLUTION RISK (bd-2c5a):\n"
|
||||
f" BEADS_WORKING_DIR={working_dir}\n"
|
||||
f" Project root: {project_root}\n"
|
||||
f" Tests should use isolated temp directories.\n"
|
||||
f" Remove BEADS_WORKING_DIR or set it to a temp directory.",
|
||||
returncode=1,
|
||||
)
|
||||
|
||||
|
||||
def pytest_runtest_setup(item):
|
||||
"""Called before each test - verify test isolation."""
|
||||
# Check if test is using bd_client fixture
|
||||
if "bd_client" in item.fixturenames:
|
||||
# Verify BEADS_DB is not set to production during test execution
|
||||
beads_db = os.environ.get("BEADS_DB", "")
|
||||
if beads_db and ".beads/beads.db" in beads_db:
|
||||
# Get temp directory
|
||||
import tempfile
|
||||
if not beads_db.startswith(tempfile.gettempdir()):
|
||||
pytest.fail(
|
||||
f"Test {item.name} is using production database (bd-2c5a):\n"
|
||||
f" BEADS_DB={beads_db}\n"
|
||||
f" This test must use a temporary database.",
|
||||
pytrace=False,
|
||||
)
|
||||
@@ -19,6 +19,11 @@ func setupTestServer(t *testing.T) (*Server, *Client, func()) {
|
||||
t.Fatalf("Failed to create temp dir: %v", err)
|
||||
}
|
||||
|
||||
// CRITICAL (bd-2c5a): Verify we're using a temp directory to prevent production pollution
|
||||
if !strings.Contains(tmpDir, os.TempDir()) {
|
||||
t.Fatalf("PRODUCTION DATABASE POLLUTION RISK (bd-2c5a): tmpDir must be in system temp directory, got: %s", tmpDir)
|
||||
}
|
||||
|
||||
// Create .beads subdirectory so findDatabaseForCwd finds THIS database, not project's
|
||||
beadsDir := filepath.Join(tmpDir, ".beads")
|
||||
if err := os.MkdirAll(beadsDir, 0750); err != nil {
|
||||
@@ -132,6 +137,11 @@ func setupTestServerIsolated(t *testing.T) (tmpDir, beadsDir, dbPath, socketPath
|
||||
t.Fatalf("Failed to create temp dir: %v", err)
|
||||
}
|
||||
|
||||
// CRITICAL (bd-2c5a): Verify we're using a temp directory to prevent production pollution
|
||||
if !strings.Contains(tmpDir, os.TempDir()) {
|
||||
t.Fatalf("PRODUCTION DATABASE POLLUTION RISK (bd-2c5a): tmpDir must be in system temp directory, got: %s", tmpDir)
|
||||
}
|
||||
|
||||
// Create .beads subdirectory so findDatabaseForCwd finds THIS database, not project's
|
||||
beadsDir = filepath.Join(tmpDir, ".beads")
|
||||
if err := os.MkdirAll(beadsDir, 0750); err != nil {
|
||||
|
||||
Reference in New Issue
Block a user