✅ Add kubernetes Folder

tests/requirements-test.txt

@@ -0,0 +1,28 @@
+# Test framework
+pytest==7.4.3
+
+# Database testing
+psycopg2-binary==2.9.9
+
+# API testing
+fastapi==0.104.1
+httpx==0.25.1
+python-multipart==0.0.6
+
+# Data processing and ML dependencies
+pandas==2.1.3
+numpy==1.26.2
+scikit-learn==1.3.2
+
+# Test client for FastAPI
+requests==2.31.0
+
+# Test client for Airflow
+airflow
+beautifulsoup4
+python-dotenv
+surprise
+mlflow
+
+# Streamlit testing
+streamlit-testing

tests/test_api_auth.py

@@ -0,0 +1,76 @@
+import pytest
+from fastapi.testclient import TestClient
+from main import app  # Assurez-vous que le chemin d'importation est correct
+
+client = TestClient(app)
+
+@pytest.fixture
+def create_user():
+    response = client.post("/auth/", json={
+        "username": "testuser",
+        "email": "[email protected]",
+        "password": "StrongPassword123!"
+    })
+    return response
+
+def test_create_user(create_user):
+    assert create_user.status_code == 201
+    assert create_user.json() is not None
+
+def test_create_user_duplicate_email(create_user):
+    response = client.post("/auth/", json={
+        "username": "anotheruser",
+        "email": "[email protected]",  # Email déjà utilisé
+        "password": "AnotherStrongPassword123!"
+    })
+    assert response.status_code == 400
+    assert response.json()["detail"] == "Email already registered"
+
+def test_login_for_access_token(create_user):
+    response = client.post("/auth/token", data={
+        "username": "testuser",
+        "password": "StrongPassword123!"
+    })
+    assert response.status_code == 200
+    assert "access_token" in response.json()
+
+def test_login_invalid_credentials():
+    response = client.post("/auth/token", data={
+        "username": "nonexistentuser",
+        "password": "WrongPassword"
+    })
+    assert response.status_code == 401
+    assert response.json()["detail"] == 'Could not validate user.'
+
+def test_validate_username():
+    # Test pour le nom d'utilisateur valide
+    valid_username = "valid_username"
+    error_message = validate_username(valid_username)
+    assert error_message is None
+
+    # Test pour le nom d'utilisateur invalide
+    invalid_username = "@invalidusername"
+    error_message = validate_username(invalid_username)
+    assert error_message == "Le nom d'utilisateur ne doit contenir que des lettres, chiffres et underscores."
+
+def test_validate_email():
+    # Test pour l'email valide
+    valid_email = "[email protected]"
+    error_message = validate_email(valid_email)
+    assert error_message is None
+
+    # Test pour l'email invalide
+    invalid_email = "invalid-email"
+    error_message = validate_email(invalid_email)
+    assert error_message == "L'adresse e-mail n'est pas valide."
+
+def test_validate_password():
+    # Test pour le mot de passe valide
+    valid_password = "StrongPassword123!"
+    error_message = validate_password(valid_password)
+    assert error_message is None
+
+    # Test pour le mot de passe invalide (trop court)
+    invalid_password = "Short1!"
+    error_message = validate_password(invalid_password)
+    assert error_message == "Le mot de passe doit contenir au moins 12 caractères."

tests/test_api_predict.py

@@ -0,0 +1,77 @@
+import pytest
+from fastapi.testclient import TestClient
+from api.main import app
+from api.auth import validate_username, validate_email, validate_password
+
+client = TestClient(app)
+
+@pytest.fixture
+def create_user():
+    response = client.post("/auth/", json={
+        "username": "testuser",
+        "email": "[email protected]",
+        "password": "StrongPassword123!"
+    })
+    return response
+
+def test_create_user(create_user):
+    assert create_user.status_code == 201
+    assert create_user.json() is not None
+
+def test_create_user_duplicate_email(create_user):
+    response = client.post("/auth/", json={
+        "username": "anotheruser",
+        "email": "[email protected]",  # Email déjà utilisé
+        "password": "AnotherStrongPassword123!"
+    })
+    assert response.status_code == 400
+    assert response.json()["detail"] == "Email already registered"
+
+def test_login_for_access_token(create_user):
+    response = client.post("/auth/token", data={
+        "username": "testuser",
+        "password": "StrongPassword123!"
+    })
+    assert response.status_code == 200
+    assert "access_token" in response.json()
+
+def test_login_invalid_credentials():
+    response = client.post("/auth/token", data={
+        "username": "nonexistentuser",
+        "password": "WrongPassword"
+    })
+    assert response.status_code == 401
+    assert response.json()["detail"] == 'Could not validate user.'
+
+def test_validate_username():
+    # Test pour le nom d'utilisateur valide
+    valid_username = "valid_username"
+    error_message = validate_username(valid_username)
+    assert error_message is None
+
+    # Test pour le nom d'utilisateur invalide
+    invalid_username = "@invalidusername"
+    error_message = validate_username(invalid_username)
+    assert error_message == "Le nom d'utilisateur ne doit contenir que des lettres, chiffres et underscores."
+
+def test_validate_email():
+    # Test pour l'email valide
+    valid_email = "[email protected]"
+    error_message = validate_email(valid_email)
+    assert error_message is None
+
+    # Test pour l'email invalide
+    invalid_email = "invalid-email"
+    error_message = validate_email(invalid_email)
+    assert error_message == "L'adresse e-mail n'est pas valide."
+
+def test_validate_password():
+    # Test pour le mot de passe valide
+    valid_password = "StrongPassword123!"
+    error_message = validate_password(valid_password)
+    assert error_message is None
+
+    # Test pour le mot de passe invalide (trop court)
+    invalid_password = "Short1!"
+    error_message = validate_password(invalid_password)
+    assert error_message == "Le mot de passe doit contenir au moins 12 caractères."

tests/test_app.py

@@ -0,0 +1,31 @@
+import streamlit as st
+from streamlit.testing import TestRunner
+from streamlit.app.utils import display_movies_grid
+from streamlit.app.pages.4_🔐_Authentification import auth_page
+from streamlit.app.pages.5_📽️_Application import app_page
+
+def test_display_movies_grid():
+    runner = TestRunner()
+    movies_info = {
+        "0": {"poster_path": "path/to/poster1.jpg", "title": "Movie 1", "vote_average": 8.5},
+        "1": {"poster_path": "path/to/poster2.jpg", "title": "Movie 2", "vote_average": 7.3},
+        "2": {"poster_path": "path/to/poster3.jpg", "title": "Movie 3", "vote_average": 9.1},
+        "3": {"poster_path": "path/to/poster4.jpg", "title": "Movie 4", "vote_average": 6.8},
+    }
+    runner.run(display_movies_grid, movies_info)
+    assert runner.get_widget("markdown").exists()
+
+def test_authentication_page():
+    runner = TestRunner()
+    runner.run(auth_page)
+    assert runner.get_widget("header").exists()
+
+def test_application_page():
+    runner = TestRunner()
+    runner.run(app_page)
+    assert runner.get_widget("markdown").exists()
+
+if __name__ == "__main__":
+    test_display_movies_grid()
+    test_authentication_page()
+    test_application_page()

tests/test_build_features.py

@@ -0,0 +1,81 @@
+
+import pytest
+import pandas as pd
+import numpy as np
+import os
+from ml.src.features.build_features import (
+    bayesienne_mean,
+    preprocessing_ratings,
+    preprocessing_movies,
+    preprocessing_links
+)
+
+@pytest.fixture
+def sample_ratings_df():
+    return pd.DataFrame({
+        'userId': [1, 1, 2, 2],
+        'movieId': [1, 2, 1, 2],
+        'rating': [4.0, 3.0, 5.0, 4.0],
+        'timestamp': [1000000, 1000001, 1000002, 1000003]
+    })
+
+@pytest.fixture
+def sample_movies_df():
+    return pd.DataFrame({
+        'movieId': [1, 2],
+        'title': ['Movie 1 (2020)', 'Movie 2 (2019)'],
+        'genres': ['Action|Adventure', 'Comedy|Drama']
+    })
+
+@pytest.fixture
+def sample_links_df():
+    return pd.DataFrame({
+        'movieId': [1, 2],
+        'imdbId': [111, 222],
+        'tmdbId': [123.0, np.nan]
+    })
+
+def test_bayesienne_mean():
+    series = pd.Series([4.0, 3.0, 5.0])
+    M = 4.0  # moyenne globale
+    C = 3.0  # nombre moyen de votes
+    result = bayesienne_mean(series, M, C)
+    assert isinstance(result, float)
+    assert 3.0 <= result <= 5.0
+
+def test_preprocessing_ratings(sample_ratings_df, tmp_path):
+    # Créer un fichier temporaire pour les tests
+    temp_file = tmp_path / "ratings.csv"
+    sample_ratings_df.to_csv(temp_file, index=False)
+
+    # Tester la fonction
+    result = preprocessing_ratings(str(temp_file))
+
+    assert isinstance(result, pd.DataFrame)
+    assert 'bayesian_mean' in result.columns
+    assert len(result) == len(sample_ratings_df)
+
+def test_preprocessing_movies(sample_movies_df, tmp_path):
+    # Créer un fichier temporaire pour les tests
+    temp_file = tmp_path / "movies.csv"
+    sample_movies_df.to_csv(temp_file, index=False)
+
+    # Tester la fonction
+    result = preprocessing_movies(str(temp_file))
+
+    assert isinstance(result, pd.DataFrame)
+    assert 'year' in result.columns
+    assert result['genres'].iloc[0] == 'Action, Adventure'
+    assert result['year'].iloc[0] == '2020'
+
+def test_preprocessing_links(sample_links_df, tmp_path):
+    # Créer un fichier temporaire pour les tests
+    temp_file = tmp_path / "links.csv"
+    sample_links_df.to_csv(temp_file, index=False)
+
+    # Tester la fonction
+    result = preprocessing_links(str(temp_file))
+
+    assert isinstance(result, pd.DataFrame)
+    assert result['tmdbId'].dtype == 'int64'
+    assert result['tmdbId'].iloc[1] == 0  # Vérifier que la valeur NaN a été remplacée par 0

tests/test_db.py

@@ -0,0 +1,36 @@
+import psycopg2
+import pytest
+from os import environ
+
+def test_database_connection():
+    try:
+        conn = psycopg2.connect(
+            dbname="test_db",
+            user="test_user",
+            password="test_password",
+            host="localhost",
+            port="5432"
+        )
+        assert conn is not None
+
+        cur = conn.cursor()
+
+        # Vérifier l'existence des tables
+        cur.execute("""
+            SELECT table_name
+            FROM information_schema.tables
+            WHERE table_schema = 'public'
+        """)
+        tables = [table[0] for table in cur.fetchall()]
+        assert 'movies' in tables
+        assert 'ratings' in tables
+        assert 'links' in tables
+        assert 'users' in tables
+
+        conn.close()
+
+    except Exception as e:
+        pytest.fail(f"Test failed: {str(e)}")
+
+if __name__ == "__main__":
+    test_database_connection()

tests/test_db_creation.sql

@@ -0,0 +1,19 @@
+SELECT table_name
+FROM information_schema.tables
+WHERE table_schema = 'public';
+
+SELECT column_name, data_type
+FROM information_schema.columns
+WHERE table_name = 'movies';
+
+SELECT column_name, data_type
+FROM information_schema.columns
+WHERE table_name = 'ratings';
+
+SELECT column_name, data_type
+FROM information_schema.columns
+WHERE table_name = 'links';
+
+SELECT column_name, data_type
+FROM information_schema.columns
+WHERE table_name = 'users';