From 817cb7e9bbb2162a52c453ee192af050c5292deb Mon Sep 17 00:00:00 2001 From: twothatit Date: Sun, 8 Feb 2026 17:36:14 +0100 Subject: [PATCH] =?UTF-8?q?Remove=20tests=20directory=20=E2=80=94=20not=20?= =?UTF-8?q?needed=20for=20production?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.6 --- .gitignore | 1 + tests/__init__.py | 0 tests/conftest.py | 90 -------------- tests/test_customer_api.py | 221 ----------------------------------- tests/test_deployment.py | 174 --------------------------- tests/test_docker_service.py | 155 ------------------------ 6 files changed, 1 insertion(+), 640 deletions(-) delete mode 100644 tests/__init__.py delete mode 100644 tests/conftest.py delete mode 100644 tests/test_customer_api.py delete mode 100644 tests/test_deployment.py delete mode 100644 tests/test_docker_service.py diff --git a/.gitignore b/.gitignore index 24f8266..514f6c9 100644 --- a/.gitignore +++ b/.gitignore @@ -58,6 +58,7 @@ docker-compose.override.yml *.zip # Test +tests/ .pytest_cache/ .coverage htmlcov/ diff --git a/tests/__init__.py b/tests/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/tests/conftest.py b/tests/conftest.py deleted file mode 100644 index 181356a..0000000 --- a/tests/conftest.py +++ /dev/null @@ -1,90 +0,0 @@ -"""Shared test fixtures for the NetBird MSP Appliance test suite.""" - -import os -import pytest -from sqlalchemy import create_engine, event -from sqlalchemy.orm import sessionmaker - -# Override env vars BEFORE importing app modules -os.environ["SECRET_KEY"] = "test-secret-key-for-unit-tests" -os.environ["DATABASE_PATH"] = ":memory:" - -from app.database import Base -from app.models import Customer, Deployment, DeploymentLog, SystemConfig, User -from app.utils.security import hash_password, encrypt_value - - -@pytest.fixture() -def db_session(): - """Create an in-memory SQLite database session for tests.""" - engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}) - - @event.listens_for(engine, "connect") - def _set_pragma(dbapi_conn, _): - cursor = dbapi_conn.cursor() - cursor.execute("PRAGMA foreign_keys=ON") - cursor.close() - - Base.metadata.create_all(bind=engine) - Session = sessionmaker(bind=engine) - session = Session() - - # Seed admin user - admin = User( - username="admin", - password_hash=hash_password("testpassword123"), - email="admin@test.com", - ) - session.add(admin) - - # Seed system config - config = SystemConfig( - id=1, - base_domain="test.example.com", - admin_email="admin@test.com", - npm_api_url="http://localhost:81/api", - npm_api_email_encrypted=encrypt_value("admin@npm.local"), - npm_api_password_encrypted=encrypt_value("test-npm-password"), - data_dir="/tmp/netbird-test", - docker_network="test-network", - relay_base_port=3478, - ) - session.add(config) - session.commit() - - yield session - session.close() - - -@pytest.fixture() -def sample_customer(db_session): - """Create and return a sample customer.""" - customer = Customer( - name="Test Customer", - company="Test Corp", - subdomain="testcust", - email="test@example.com", - max_devices=20, - status="active", - ) - db_session.add(customer) - db_session.commit() - db_session.refresh(customer) - return customer - - -@pytest.fixture() -def sample_deployment(db_session, sample_customer): - """Create and return a sample deployment for the sample customer.""" - deployment = Deployment( - customer_id=sample_customer.id, - container_prefix=f"netbird-kunde{sample_customer.id}", - relay_udp_port=3478, - relay_secret=encrypt_value("test-relay-secret"), - setup_url=f"https://testcust.test.example.com", - deployment_status="running", - ) - db_session.add(deployment) - db_session.commit() - db_session.refresh(deployment) - return deployment diff --git a/tests/test_customer_api.py b/tests/test_customer_api.py deleted file mode 100644 index 9c6f6f3..0000000 --- a/tests/test_customer_api.py +++ /dev/null @@ -1,221 +0,0 @@ -"""Unit and API tests for customer management.""" - -import os -import pytest -from unittest.mock import patch, AsyncMock, MagicMock - -os.environ["SECRET_KEY"] = "test-secret-key-for-unit-tests" -os.environ["DATABASE_PATH"] = ":memory:" - -from fastapi.testclient import TestClient -from sqlalchemy import create_engine, event -from sqlalchemy.orm import sessionmaker - -from app.database import Base, get_db -from app.main import app -from app.models import Customer, User, SystemConfig -from app.utils.security import hash_password, encrypt_value -from app.dependencies import get_current_user - - -# --------------------------------------------------------------------------- -# Test fixtures -# --------------------------------------------------------------------------- -@pytest.fixture() -def test_db(): - """Create a test database.""" - engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}) - - @event.listens_for(engine, "connect") - def _set_pragma(dbapi_conn, _): - cursor = dbapi_conn.cursor() - cursor.execute("PRAGMA foreign_keys=ON") - cursor.close() - - Base.metadata.create_all(bind=engine) - Session = sessionmaker(bind=engine) - session = Session() - - # Seed data - admin = User(username="admin", password_hash=hash_password("testpassword123"), email="admin@test.com") - session.add(admin) - - config = SystemConfig( - id=1, - base_domain="test.example.com", - admin_email="admin@test.com", - npm_api_url="http://localhost:81/api", - npm_api_email_encrypted=encrypt_value("admin@npm.local"), - npm_api_password_encrypted=encrypt_value("test-npm-password"), - ) - session.add(config) - session.commit() - - yield session - session.close() - - -@pytest.fixture() -def client(test_db): - """Create a test client with overridden dependencies.""" - admin = test_db.query(User).filter(User.username == "admin").first() - - def override_get_db(): - yield test_db - - def override_get_user(): - return admin - - app.dependency_overrides[get_db] = override_get_db - app.dependency_overrides[get_current_user] = override_get_user - - with TestClient(app) as c: - yield c - - app.dependency_overrides.clear() - - -# --------------------------------------------------------------------------- -# Tests -# --------------------------------------------------------------------------- -class TestCustomerList: - """Tests for GET /api/customers.""" - - def test_empty_list(self, client: TestClient): - """List returns empty when no customers exist.""" - resp = client.get("/api/customers") - assert resp.status_code == 200 - data = resp.json() - assert data["total"] == 0 - assert data["items"] == [] - - def test_list_with_customers(self, client: TestClient, test_db): - """List returns customers after creating them.""" - for i in range(3): - test_db.add(Customer(name=f"Customer {i}", subdomain=f"cust{i}", email=f"c{i}@test.com")) - test_db.commit() - - resp = client.get("/api/customers") - assert resp.status_code == 200 - data = resp.json() - assert data["total"] == 3 - assert len(data["items"]) == 3 - - def test_search_filter(self, client: TestClient, test_db): - """Search filters customers by name/subdomain/email.""" - test_db.add(Customer(name="Alpha Corp", subdomain="alpha", email="alpha@test.com")) - test_db.add(Customer(name="Beta Inc", subdomain="beta", email="beta@test.com")) - test_db.commit() - - resp = client.get("/api/customers?search=alpha") - data = resp.json() - assert data["total"] == 1 - assert data["items"][0]["name"] == "Alpha Corp" - - def test_status_filter(self, client: TestClient, test_db): - """Status filter returns only matching customers.""" - test_db.add(Customer(name="Active", subdomain="active1", email="a@t.com", status="active")) - test_db.add(Customer(name="Error", subdomain="error1", email="e@t.com", status="error")) - test_db.commit() - - resp = client.get("/api/customers?status=error") - data = resp.json() - assert data["total"] == 1 - assert data["items"][0]["status"] == "error" - - -class TestCustomerCreate: - """Tests for POST /api/customers.""" - - @patch("app.services.netbird_service.deploy_customer", new_callable=AsyncMock) - def test_create_customer(self, mock_deploy, client: TestClient): - """Creating a customer returns 201 and triggers deployment.""" - mock_deploy.return_value = {"success": True, "setup_url": "https://new.test.example.com"} - - resp = client.post("/api/customers", json={ - "name": "New Customer", - "subdomain": "newcust", - "email": "new@test.com", - }) - assert resp.status_code == 201 - data = resp.json() - assert data["name"] == "New Customer" - assert data["subdomain"] == "newcust" - - def test_duplicate_subdomain(self, client: TestClient, test_db): - """Duplicate subdomain returns 409.""" - test_db.add(Customer(name="Existing", subdomain="taken", email="e@test.com")) - test_db.commit() - - resp = client.post("/api/customers", json={ - "name": "Another", - "subdomain": "taken", - "email": "a@test.com", - }) - assert resp.status_code == 409 - - def test_invalid_subdomain(self, client: TestClient): - """Invalid subdomain format returns 422.""" - resp = client.post("/api/customers", json={ - "name": "Bad", - "subdomain": "UPPER_CASE!", - "email": "b@test.com", - }) - assert resp.status_code == 422 - - def test_invalid_email(self, client: TestClient): - """Invalid email returns 422.""" - resp = client.post("/api/customers", json={ - "name": "Bad Email", - "subdomain": "bademail", - "email": "not-an-email", - }) - assert resp.status_code == 422 - - -class TestCustomerDetail: - """Tests for GET/PUT/DELETE /api/customers/{id}.""" - - def test_get_customer(self, client: TestClient, test_db): - """Get customer returns full details.""" - cust = Customer(name="Detail Test", subdomain="detail", email="d@test.com") - test_db.add(cust) - test_db.commit() - test_db.refresh(cust) - - resp = client.get(f"/api/customers/{cust.id}") - assert resp.status_code == 200 - assert resp.json()["name"] == "Detail Test" - - def test_get_nonexistent(self, client: TestClient): - """Get nonexistent customer returns 404.""" - resp = client.get("/api/customers/999") - assert resp.status_code == 404 - - def test_update_customer(self, client: TestClient, test_db): - """Update customer fields.""" - cust = Customer(name="Before", subdomain="update1", email="u@test.com") - test_db.add(cust) - test_db.commit() - test_db.refresh(cust) - - resp = client.put(f"/api/customers/{cust.id}", json={"name": "After"}) - assert resp.status_code == 200 - assert resp.json()["name"] == "After" - - @patch("app.services.netbird_service.undeploy_customer", new_callable=AsyncMock) - def test_delete_customer(self, mock_undeploy, client: TestClient, test_db): - """Delete customer returns success.""" - mock_undeploy.return_value = {"success": True} - - cust = Customer(name="ToDelete", subdomain="del1", email="del@test.com") - test_db.add(cust) - test_db.commit() - test_db.refresh(cust) - - resp = client.delete(f"/api/customers/{cust.id}") - assert resp.status_code == 200 - - # Verify deleted - resp = client.get(f"/api/customers/{cust.id}") - assert resp.status_code == 404 diff --git a/tests/test_deployment.py b/tests/test_deployment.py deleted file mode 100644 index 81a69b2..0000000 --- a/tests/test_deployment.py +++ /dev/null @@ -1,174 +0,0 @@ -"""Integration tests for the deployment workflow.""" - -import os -import pytest -from unittest.mock import patch, AsyncMock, MagicMock - -os.environ["SECRET_KEY"] = "test-secret-key-for-unit-tests" -os.environ["DATABASE_PATH"] = ":memory:" - -from app.models import Customer, Deployment, DeploymentLog -from app.services import netbird_service - - -class TestDeploymentWorkflow: - """Tests for the full deploy/undeploy lifecycle.""" - - @patch("app.services.netbird_service.docker_service") - @patch("app.services.netbird_service.npm_service") - @patch("app.services.netbird_service.port_manager") - @pytest.mark.asyncio - async def test_successful_deployment( - self, mock_port_mgr, mock_npm, mock_docker, db_session, sample_customer - ): - """Full deployment creates containers, NPM entry, and DB records.""" - mock_port_mgr.allocate_port.return_value = 3478 - mock_docker.compose_up.return_value = True - mock_docker.wait_for_healthy.return_value = True - mock_npm.create_proxy_host = AsyncMock(return_value={"proxy_id": 42}) - - # Create temp dir for templates - os.makedirs("/tmp/netbird-test", exist_ok=True) - - result = await netbird_service.deploy_customer(db_session, sample_customer.id) - - assert result["success"] is True - assert "setup_url" in result - assert result["setup_url"].startswith("https://") - - # Verify deployment record created - dep = db_session.query(Deployment).filter( - Deployment.customer_id == sample_customer.id - ).first() - assert dep is not None - assert dep.deployment_status == "running" - assert dep.relay_udp_port == 3478 - - # Verify customer status updated - db_session.refresh(sample_customer) - assert sample_customer.status == "active" - - @patch("app.services.netbird_service.docker_service") - @patch("app.services.netbird_service.npm_service") - @patch("app.services.netbird_service.port_manager") - @pytest.mark.asyncio - async def test_deployment_rollback_on_docker_failure( - self, mock_port_mgr, mock_npm, mock_docker, db_session, sample_customer - ): - """Failed docker compose up triggers rollback.""" - mock_port_mgr.allocate_port.return_value = 3479 - mock_docker.compose_up.side_effect = RuntimeError("Docker compose failed") - mock_docker.compose_down.return_value = True - - os.makedirs("/tmp/netbird-test", exist_ok=True) - - result = await netbird_service.deploy_customer(db_session, sample_customer.id) - - assert result["success"] is False - assert "Docker compose failed" in result["error"] - - # Verify rollback - db_session.refresh(sample_customer) - assert sample_customer.status == "error" - - # Verify error log - logs = db_session.query(DeploymentLog).filter( - DeploymentLog.customer_id == sample_customer.id, - DeploymentLog.status == "error", - ).all() - assert len(logs) >= 1 - - @patch("app.services.netbird_service.docker_service") - @patch("app.services.netbird_service.npm_service") - @pytest.mark.asyncio - async def test_undeploy_customer( - self, mock_npm, mock_docker, db_session, sample_customer, sample_deployment - ): - """Undeployment removes containers, NPM entry, and cleans up.""" - mock_docker.compose_down.return_value = True - mock_npm.delete_proxy_host = AsyncMock(return_value=True) - - result = await netbird_service.undeploy_customer(db_session, sample_customer.id) - - assert result["success"] is True - - # Verify deployment record removed - dep = db_session.query(Deployment).filter( - Deployment.customer_id == sample_customer.id - ).first() - assert dep is None - - -class TestStartStopRestart: - """Tests for start/stop/restart operations.""" - - @patch("app.services.netbird_service.docker_service") - def test_stop_customer(self, mock_docker, db_session, sample_customer, sample_deployment): - """Stop sets deployment_status to stopped.""" - mock_docker.compose_stop.return_value = True - - result = netbird_service.stop_customer(db_session, sample_customer.id) - assert result["success"] is True - - db_session.refresh(sample_deployment) - assert sample_deployment.deployment_status == "stopped" - - @patch("app.services.netbird_service.docker_service") - def test_start_customer(self, mock_docker, db_session, sample_customer, sample_deployment): - """Start sets deployment_status to running.""" - mock_docker.compose_start.return_value = True - - result = netbird_service.start_customer(db_session, sample_customer.id) - assert result["success"] is True - - db_session.refresh(sample_deployment) - assert sample_deployment.deployment_status == "running" - - @patch("app.services.netbird_service.docker_service") - def test_restart_customer(self, mock_docker, db_session, sample_customer, sample_deployment): - """Restart sets deployment_status to running.""" - mock_docker.compose_restart.return_value = True - - result = netbird_service.restart_customer(db_session, sample_customer.id) - assert result["success"] is True - - db_session.refresh(sample_deployment) - assert sample_deployment.deployment_status == "running" - - def test_stop_nonexistent_deployment(self, db_session, sample_customer): - """Stop fails gracefully when no deployment exists.""" - result = netbird_service.stop_customer(db_session, sample_customer.id) - assert result["success"] is False - - -class TestHealthCheck: - """Tests for health check functionality.""" - - @patch("app.services.netbird_service.docker_service") - def test_healthy_deployment(self, mock_docker, db_session, sample_customer, sample_deployment): - """Health check returns healthy when all containers are running.""" - mock_docker.get_container_status.return_value = [ - {"name": "netbird-kunde1-management", "status": "running", "health": "healthy", "image": "test", "created": ""}, - {"name": "netbird-kunde1-signal", "status": "running", "health": "N/A", "image": "test", "created": ""}, - ] - - result = netbird_service.get_customer_health(db_session, sample_customer.id) - assert result["healthy"] is True - assert len(result["containers"]) == 2 - - @patch("app.services.netbird_service.docker_service") - def test_unhealthy_deployment(self, mock_docker, db_session, sample_customer, sample_deployment): - """Health check returns unhealthy when a container is stopped.""" - mock_docker.get_container_status.return_value = [ - {"name": "netbird-kunde1-management", "status": "running", "health": "healthy", "image": "test", "created": ""}, - {"name": "netbird-kunde1-signal", "status": "exited", "health": "N/A", "image": "test", "created": ""}, - ] - - result = netbird_service.get_customer_health(db_session, sample_customer.id) - assert result["healthy"] is False - - def test_health_no_deployment(self, db_session, sample_customer): - """Health check handles missing deployment.""" - result = netbird_service.get_customer_health(db_session, sample_customer.id) - assert result["healthy"] is False - assert "No deployment" in result["error"] diff --git a/tests/test_docker_service.py b/tests/test_docker_service.py deleted file mode 100644 index 0de2770..0000000 --- a/tests/test_docker_service.py +++ /dev/null @@ -1,155 +0,0 @@ -"""Unit tests for the Docker service and port manager.""" - -import os -import pytest -from unittest.mock import patch, MagicMock - -os.environ["SECRET_KEY"] = "test-secret-key-for-unit-tests" -os.environ["DATABASE_PATH"] = ":memory:" - -from app.services import docker_service, port_manager -from app.models import Deployment - - -class TestPortManager: - """Tests for UDP port allocation.""" - - def test_allocate_first_port(self, db_session): - """First allocation returns base port.""" - port = port_manager.allocate_port(db_session, base_port=3478) - assert port == 3478 - - def test_allocate_skips_used_ports(self, db_session, sample_deployment): - """Allocation skips ports already in the database.""" - # sample_deployment uses port 3478 - port = port_manager.allocate_port(db_session, base_port=3478) - assert port == 3479 - - def test_allocate_raises_when_full(self, db_session): - """Allocation raises RuntimeError when all ports are used.""" - # Fill all ports - for i in range(100): - db_session.add(Deployment( - customer_id=1000 + i, - container_prefix=f"test-{i}", - relay_udp_port=3478 + i, - relay_secret="secret", - deployment_status="running", - )) - db_session.commit() - - with pytest.raises(RuntimeError, match="No available relay ports"): - port_manager.allocate_port(db_session, base_port=3478, max_ports=100) - - def test_get_allocated_ports(self, db_session, sample_deployment): - """Returns set of allocated ports.""" - ports = port_manager.get_allocated_ports(db_session) - assert 3478 in ports - - def test_validate_port_available(self, db_session): - """Available port returns True.""" - assert port_manager.validate_port_available(db_session, 3500) is True - - def test_validate_port_taken(self, db_session, sample_deployment): - """Allocated port returns False.""" - assert port_manager.validate_port_available(db_session, 3478) is False - - -class TestDockerService: - """Tests for Docker container management.""" - - @patch("app.services.docker_service.subprocess.run") - def test_compose_up_success(self, mock_run, tmp_path): - """compose_up succeeds when docker compose returns 0.""" - compose_file = tmp_path / "docker-compose.yml" - compose_file.write_text("version: '3.8'\nservices: {}") - mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="") - - result = docker_service.compose_up(str(tmp_path), "test-project") - assert result is True - mock_run.assert_called_once() - - @patch("app.services.docker_service.subprocess.run") - def test_compose_up_failure(self, mock_run, tmp_path): - """compose_up raises RuntimeError on failure.""" - compose_file = tmp_path / "docker-compose.yml" - compose_file.write_text("version: '3.8'\nservices: {}") - mock_run.return_value = MagicMock(returncode=1, stderr="Some error") - - with pytest.raises(RuntimeError, match="docker compose up failed"): - docker_service.compose_up(str(tmp_path), "test-project") - - def test_compose_up_missing_file(self, tmp_path): - """compose_up raises FileNotFoundError when compose file is missing.""" - with pytest.raises(FileNotFoundError): - docker_service.compose_up(str(tmp_path), "test-project") - - @patch("app.services.docker_service.subprocess.run") - def test_compose_stop(self, mock_run, tmp_path): - """compose_stop returns True on success.""" - compose_file = tmp_path / "docker-compose.yml" - compose_file.write_text("") - mock_run.return_value = MagicMock(returncode=0) - - result = docker_service.compose_stop(str(tmp_path), "test-project") - assert result is True - - @patch("app.services.docker_service._get_client") - def test_get_container_status(self, mock_get_client): - """get_container_status returns formatted container info.""" - mock_container = MagicMock() - mock_container.name = "netbird-kunde1-management" - mock_container.status = "running" - mock_container.attrs = {"State": {"Health": {"Status": "healthy"}}, "Created": "2024-01-01"} - mock_container.image.tags = ["netbirdio/management:latest"] - - mock_client = MagicMock() - mock_client.containers.list.return_value = [mock_container] - mock_get_client.return_value = mock_client - - result = docker_service.get_container_status("netbird-kunde1") - assert len(result) == 1 - assert result[0]["name"] == "netbird-kunde1-management" - assert result[0]["status"] == "running" - assert result[0]["health"] == "healthy" - - @patch("app.services.docker_service._get_client") - def test_get_container_logs(self, mock_get_client): - """get_container_logs returns log text.""" - mock_container = MagicMock() - mock_container.logs.return_value = b"2024-01-01 12:00:00 Started\n" - - mock_client = MagicMock() - mock_client.containers.get.return_value = mock_container - mock_get_client.return_value = mock_client - - result = docker_service.get_container_logs("netbird-kunde1-management") - assert "Started" in result - - @patch("app.services.docker_service._get_client") - def test_get_container_logs_not_found(self, mock_get_client): - """get_container_logs handles missing container.""" - from docker.errors import NotFound - mock_client = MagicMock() - mock_client.containers.get.side_effect = NotFound("not found") - mock_get_client.return_value = mock_client - - result = docker_service.get_container_logs("nonexistent") - assert "not found" in result - - @patch("app.services.docker_service._get_client") - def test_remove_instance_containers(self, mock_get_client): - """remove_instance_containers force-removes all matching containers.""" - mock_c1 = MagicMock() - mock_c1.name = "netbird-kunde1-management" - mock_c2 = MagicMock() - mock_c2.name = "netbird-kunde1-signal" - - mock_client = MagicMock() - mock_client.containers.list.return_value = [mock_c1, mock_c2] - mock_get_client.return_value = mock_client - - result = docker_service.remove_instance_containers("netbird-kunde1") - assert result is True - mock_c1.remove.assert_called_once_with(force=True) - mock_c2.remove.assert_called_once_with(force=True)