ipwatch

tests/__init__.py

@@ -0,0 +1 @@
+# Tests IPWatch

tests/test_api.py

@@ -0,0 +1,123 @@
+"""
+Tests pour les endpoints API
+"""
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from backend.app.main import app
+from backend.app.core.database import Base, get_db
+from backend.app.models.ip import IP
+
+
+# Setup DB de test
+@pytest.fixture
+def test_db():
+    """Fixture base de données de test"""
+    engine = create_engine("sqlite:///:memory:")
+    Base.metadata.create_all(engine)
+    TestingSessionLocal = sessionmaker(bind=engine)
+    return TestingSessionLocal
+
+
+@pytest.fixture
+def client(test_db):
+    """Fixture client de test"""
+    def override_get_db():
+        db = test_db()
+        try:
+            yield db
+        finally:
+            db.close()
+
+    app.dependency_overrides[get_db] = override_get_db
+    return TestClient(app)
+
+
+class TestAPIEndpoints:
+    """Tests pour les endpoints API"""
+
+    def test_root_endpoint(self, client):
+        """Test endpoint racine"""
+        response = client.get("/")
+        assert response.status_code == 200
+        data = response.json()
+        assert "name" in data
+        assert data["name"] == "IPWatch API"
+
+    def test_health_check(self, client):
+        """Test health check"""
+        response = client.get("/health")
+        assert response.status_code == 200
+        data = response.json()
+        assert "status" in data
+        assert data["status"] == "healthy"
+
+    def test_get_all_ips_empty(self, client):
+        """Test récupération IPs (vide)"""
+        response = client.get("/api/ips/")
+        assert response.status_code == 200
+        data = response.json()
+        assert isinstance(data, list)
+        assert len(data) == 0
+
+    def test_get_stats_empty(self, client):
+        """Test stats avec DB vide"""
+        response = client.get("/api/ips/stats/summary")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["total"] == 0
+        assert data["online"] == 0
+        assert data["offline"] == 0
+
+    def test_get_ip_not_found(self, client):
+        """Test récupération IP inexistante"""
+        response = client.get("/api/ips/192.168.1.100")
+        assert response.status_code == 404
+
+    def test_update_ip(self, client, test_db):
+        """Test mise à jour IP"""
+        # Créer d'abord une IP
+        db = test_db()
+        ip = IP(
+            ip="192.168.1.100",
+            name="Test",
+            known=False,
+            last_status="online"
+        )
+        db.add(ip)
+        db.commit()
+        db.close()
+
+        # Mettre à jour via API
+        update_data = {
+            "name": "Updated Name",
+            "known": True,
+            "location": "Bureau"
+        }
+
+        response = client.put("/api/ips/192.168.1.100", json=update_data)
+        assert response.status_code == 200
+
+        data = response.json()
+        assert data["name"] == "Updated Name"
+        assert data["known"] is True
+        assert data["location"] == "Bureau"
+
+    def test_delete_ip(self, client, test_db):
+        """Test suppression IP"""
+        # Créer une IP
+        db = test_db()
+        ip = IP(ip="192.168.1.101", last_status="online")
+        db.add(ip)
+        db.commit()
+        db.close()
+
+        # Supprimer via API
+        response = client.delete("/api/ips/192.168.1.101")
+        assert response.status_code == 200
+
+        # Vérifier suppression
+        response = client.get("/api/ips/192.168.1.101")
+        assert response.status_code == 404

tests/test_models.py

@@ -0,0 +1,134 @@
+"""
+Tests pour les modèles SQLAlchemy
+"""
+import pytest
+from datetime import datetime
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from backend.app.core.database import Base
+from backend.app.models.ip import IP, IPHistory
+
+
+class TestSQLAlchemyModels:
+    """Tests pour les modèles de données"""
+
+    @pytest.fixture
+    def db_session(self):
+        """Fixture session DB en mémoire"""
+        # Créer une DB SQLite en mémoire
+        engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(engine)
+
+        Session = sessionmaker(bind=engine)
+        session = Session()
+
+        yield session
+
+        session.close()
+
+    def test_create_ip(self, db_session):
+        """Test création d'une IP"""
+        ip = IP(
+            ip="192.168.1.100",
+            name="Test Server",
+            known=True,
+            location="Bureau",
+            host="Serveur",
+            last_status="online",
+            mac="00:11:22:33:44:55",
+            vendor="Dell",
+            open_ports=[22, 80, 443]
+        )
+
+        db_session.add(ip)
+        db_session.commit()
+
+        # Vérifier la création
+        retrieved = db_session.query(IP).filter(IP.ip == "192.168.1.100").first()
+        assert retrieved is not None
+        assert retrieved.name == "Test Server"
+        assert retrieved.known is True
+        assert retrieved.last_status == "online"
+        assert len(retrieved.open_ports) == 3
+
+    def test_create_ip_history(self, db_session):
+        """Test création d'historique IP"""
+        # Créer d'abord une IP
+        ip = IP(
+            ip="192.168.1.101",
+            last_status="online"
+        )
+        db_session.add(ip)
+        db_session.commit()
+
+        # Créer entrée historique
+        history = IPHistory(
+            ip="192.168.1.101",
+            timestamp=datetime.now(),
+            status="online",
+            open_ports=[80, 443]
+        )
+
+        db_session.add(history)
+        db_session.commit()
+
+        # Vérifier
+        retrieved = db_session.query(IPHistory).filter(
+            IPHistory.ip == "192.168.1.101"
+        ).first()
+
+        assert retrieved is not None
+        assert retrieved.status == "online"
+        assert len(retrieved.open_ports) == 2
+
+    def test_ip_history_relationship(self, db_session):
+        """Test relation IP <-> IPHistory"""
+        # Créer une IP
+        ip = IP(
+            ip="192.168.1.102",
+            last_status="online"
+        )
+        db_session.add(ip)
+        db_session.commit()
+
+        # Créer plusieurs entrées historiques
+        for i in range(5):
+            history = IPHistory(
+                ip="192.168.1.102",
+                status="online" if i % 2 == 0 else "offline",
+                open_ports=[]
+            )
+            db_session.add(history)
+
+        db_session.commit()
+
+        # Vérifier la relation
+        ip = db_session.query(IP).filter(IP.ip == "192.168.1.102").first()
+        assert len(ip.history) == 5
+
+    def test_cascade_delete(self, db_session):
+        """Test suppression en cascade"""
+        # Créer IP + historique
+        ip = IP(ip="192.168.1.103", last_status="online")
+        db_session.add(ip)
+        db_session.commit()
+
+        history = IPHistory(
+            ip="192.168.1.103",
+            status="online",
+            open_ports=[]
+        )
+        db_session.add(history)
+        db_session.commit()
+
+        # Supprimer l'IP
+        db_session.delete(ip)
+        db_session.commit()
+
+        # Vérifier que l'historique est supprimé aussi
+        history_count = db_session.query(IPHistory).filter(
+            IPHistory.ip == "192.168.1.103"
+        ).count()
+
+        assert history_count == 0

tests/test_network.py

@@ -0,0 +1,98 @@
+"""
+Tests unitaires pour les modules réseau
+Basé sur tests-backend.md
+"""
+import pytest
+import asyncio
+from backend.app.services.network import NetworkScanner
+
+
+class TestNetworkScanner:
+    """Tests pour le scanner réseau"""
+
+    @pytest.fixture
+    def scanner(self):
+        """Fixture scanner avec réseau de test"""
+        return NetworkScanner(cidr="192.168.1.0/24", timeout=1.0)
+
+    def test_generate_ip_list(self, scanner):
+        """Test génération liste IP depuis CIDR"""
+        ip_list = scanner.generate_ip_list()
+
+        # Vérifier le nombre d'IPs (254 pour un /24)
+        assert len(ip_list) == 254
+
+        # Vérifier format
+        assert "192.168.1.1" in ip_list
+        assert "192.168.1.254" in ip_list
+        assert "192.168.1.0" not in ip_list  # Adresse réseau exclue
+        assert "192.168.1.255" not in ip_list  # Broadcast exclu
+
+    @pytest.mark.asyncio
+    async def test_ping(self, scanner):
+        """Test fonction ping"""
+        # Ping localhost (devrait marcher)
+        result = await scanner.ping("127.0.0.1")
+        assert result is True
+
+        # Ping IP improbable (devrait échouer rapidement)
+        result = await scanner.ping("192.0.2.1")
+        assert result is False
+
+    @pytest.mark.asyncio
+    async def test_ping_parallel(self, scanner):
+        """Test ping parallélisé"""
+        ip_list = ["127.0.0.1", "192.0.2.1", "192.0.2.2"]
+
+        results = await scanner.ping_parallel(ip_list, max_concurrent=10)
+
+        # Vérifier que tous les résultats sont présents
+        assert len(results) == 3
+        assert "127.0.0.1" in results
+        assert results["127.0.0.1"] is True
+
+    def test_classification(self, scanner):
+        """Test classification d'état IP"""
+        # IP en ligne + connue
+        status = scanner.classify_ip_status(is_online=True, is_known=True)
+        assert status == "online"
+
+        # IP hors ligne + connue
+        status = scanner.classify_ip_status(is_online=False, is_known=True)
+        assert status == "offline"
+
+        # IP en ligne + inconnue
+        status = scanner.classify_ip_status(is_online=True, is_known=False)
+        assert status == "online"
+
+        # IP hors ligne + inconnue
+        status = scanner.classify_ip_status(is_online=False, is_known=False)
+        assert status == "offline"
+
+    @pytest.mark.asyncio
+    async def test_port_scan(self, scanner):
+        """Test scan de ports"""
+        # Scanner des ports communs sur localhost
+        ports = [22, 80, 443, 9999]  # 9999 probablement fermé
+
+        open_ports = await scanner.scan_ports("127.0.0.1", ports)
+
+        # Au moins vérifier que la fonction retourne une liste
+        assert isinstance(open_ports, list)
+
+        # Tous les ports retournés doivent être dans la liste demandée
+        for port in open_ports:
+            assert port in ports
+
+    def test_get_mac_vendor(self, scanner):
+        """Test lookup fabricant MAC"""
+        # Tester avec des MACs connus
+        vendor = scanner._get_mac_vendor("00:0C:29:XX:XX:XX")
+        assert vendor == "VMware"
+
+        vendor = scanner._get_mac_vendor("B8:27:EB:XX:XX:XX")
+        assert vendor == "Raspberry Pi"
+
+        # MAC inconnu
+        vendor = scanner._get_mac_vendor("AA:BB:CC:DD:EE:FF")
+        assert vendor == "Unknown"

tests/test_scheduler.py

@@ -0,0 +1,76 @@
+"""
+Tests pour le scheduler APScheduler
+"""
+import pytest
+import asyncio
+from backend.app.services.scheduler import ScanScheduler
+
+
+class TestScheduler:
+    """Tests pour le scheduler"""
+
+    @pytest.fixture
+    def scheduler(self):
+        """Fixture scheduler"""
+        sched = ScanScheduler()
+        yield sched
+        if sched.is_running:
+            sched.stop()
+
+    def test_scheduler_start_stop(self, scheduler):
+        """Test démarrage/arrêt du scheduler"""
+        assert scheduler.is_running is False
+
+        scheduler.start()
+        assert scheduler.is_running is True
+
+        scheduler.stop()
+        assert scheduler.is_running is False
+
+    def test_add_ping_scan_job(self, scheduler):
+        """Test ajout tâche ping scan"""
+        scheduler.start()
+
+        async def dummy_scan():
+            pass
+
+        scheduler.add_ping_scan_job(dummy_scan, interval_seconds=60)
+
+        jobs = scheduler.get_jobs()
+        job_ids = [job.id for job in jobs]
+
+        assert 'ping_scan' in job_ids
+
+    def test_add_port_scan_job(self, scheduler):
+        """Test ajout tâche port scan"""
+        scheduler.start()
+
+        async def dummy_scan():
+            pass
+
+        scheduler.add_port_scan_job(dummy_scan, interval_seconds=300)
+
+        jobs = scheduler.get_jobs()
+        job_ids = [job.id for job in jobs]
+
+        assert 'port_scan' in job_ids
+
+    def test_remove_job(self, scheduler):
+        """Test suppression de tâche"""
+        scheduler.start()
+
+        async def dummy_scan():
+            pass
+
+        scheduler.add_ping_scan_job(dummy_scan, interval_seconds=60)
+
+        # Vérifier présence
+        jobs = scheduler.get_jobs()
+        assert len(jobs) == 1
+
+        # Supprimer
+        scheduler.remove_job('ping_scan')
+
+        # Vérifier absence
+        jobs = scheduler.get_jobs()
+        assert len(jobs) == 0