diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
new file mode 100644
index 0000000..a614091
--- /dev/null
+++ b/.github/workflows/tests.yml
@@ -0,0 +1,41 @@
+name: tests
+
+on:
+  push:
+    branches: [ "**" ]
+
+jobs:
+  pytest:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.9", "3.10", "3.11"]
+
+    steps:
+      - name: Check out repo
+        uses: actions/checkout@v4
+
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+          cache: "pip"
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          # If you have dev deps in requirements-dev.txt, install that; else just requirements.txt
+          if [ -f requirements-dev.txt ]; then pip install -r requirements-dev.txt; fi
+          if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+          # Install your package in editable mode if you use src/ layout (optional but nice)
+          if [ -f pyproject.toml ] || [ -f setup.cfg ] || [ -f setup.py ]; then pip install -e .; fi
+          # Always install pytest (in case it’s not in the reqs)
+          pip install pytest
+
+      - name: Run tests
+        env:
+          # Make sure tests never hit your real API
+          BASE_URL: "http://testserver"
+        run: |
+          pytest -q
\ No newline at end of file
diff --git a/data_ingestion/__init__.py b/data_ingestion/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/data_ingestion/fetch_data.py b/data_ingestion/fetch_data.py
new file mode 100644
index 0000000..67aa691
--- /dev/null
+++ b/data_ingestion/fetch_data.py
@@ -0,0 +1,37 @@
+import requests
+import os
+import datetime
+from urllib3.exceptions import NotOpenSSLWarning
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
+
+BASE_URL = os.getenv("BASE_URL", "http://127.0.0.1:8000")
+
+def retry_mechanism(total = 3, backoff = 1):
+    session = requests.Session()
+    retry = Retry(
+        total=total,
+        backoff_factor=backoff,
+        status_forcelist=(500, 502, 503, 504, 429),
+    )
+    adapter = HTTPAdapter(max_retries=retry)
+    session.mount("http://", adapter)
+
+    return session
+
+def iter_pages(endpoint: str, size):
+    sess = retry_mechanism()
+    page = 1
+    while True:
+        resp = sess.get(f"{BASE_URL.rstrip('/')}/{endpoint.lstrip('/')}",
+                        params={"page": page, "size": size}, timeout=15)
+        resp.raise_for_status()
+        data = resp.json()
+        yield data # This will yield (stream) each page of data as a dictionary. It streams pages and bulk upserts per page, which minimizes memory.
+        if page >= data.get("pages", page):
+            break
+        page += 1
+
+if __name__ == "__main__":
+    main()
+
diff --git a/data_ingestion/load_data.py b/data_ingestion/load_data.py
new file mode 100644
index 0000000..8318650
--- /dev/null
+++ b/data_ingestion/load_data.py
@@ -0,0 +1,130 @@
+import json
+from datetime import datetime
+from pathlib import Path
+from typing import Dict, Any, Iterable
+
+from data_ingestion.fetch_data import retry_mechanism, iter_pages
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+DATA_DIR = REPO_ROOT / "data_store"
+
+
+PAGE_SIZE = 100
+
+TABLES = {
+    "tracks":         {"pk": "id",       "file": DATA_DIR / "tracks.json"},
+    "users":          {"pk": "id",       "file": DATA_DIR / "users.json"},
+    "listen_history": {"pk": "user_id",  "file": DATA_DIR / "listen_history.json"},
+}
+
+WATERMARK_FILE = DATA_DIR / "watermark.json" # This file stores the last processed row for each table. Will be used from incremental loads.
+
+# -- functions for local storage -- #
+
+def ensure_dirs():
+    """
+    Ensure that the data directory exists.
+    """
+    DATA_DIR.mkdir(parents=True, exist_ok=True)
+
+def load_json(path, default):
+    if not path.exists():
+        return default # empty dict
+    with path.open("r", encoding="utf-8") as f:
+        return json.load(f)
+
+def save_json(path, obj):
+    tmp = path.with_suffix(path.suffix + ".tmp")
+    with tmp.open("w", encoding="utf-8") as f:
+        json.dump(obj, f, ensure_ascii=False, indent=2)
+    tmp.replace(path)
+
+
+def to_dt(s):
+    try:
+        return datetime.fromisoformat(s) if s else None
+    except Exception:
+        return None
+
+# -- watermarks -- #
+
+def get_watermark(table_name):
+    w = load_json(WATERMARK_FILE, {})
+    value = w.get(table_name)
+    return to_dt(value) if value else None 
+
+def set_watermark(table_name, value: datetime):
+    if value is None:
+        return
+    w = load_json(WATERMARK_FILE, {})
+    w[table_name] = value.isoformat()
+    save_json(WATERMARK_FILE, w)
+
+# -- data ingestion -- #
+
+def bulk_insert(table_name, rows: Iterable[Dict[str, Any]]):
+    """
+    Store data in a json file (one per table), keyed by the primary key.
+    If key exists, it will be overwritten if updated_at is newer.
+    """
+    ensure_dirs()
+
+    table = TABLES[table_name]
+    file_path = table["file"]
+    pk = table["pk"]
+
+    stored = load_json(file_path, {})
+    changed = False
+
+    for row in rows:
+        key = row.get(pk) 
+        if key is None:
+            continue
+        if key not in stored:
+            stored[key] = row
+            changed = True
+        else:
+            current_updated_at = to_dt(stored[key].get("updated_at"))
+            new_updated_at = to_dt(row.get("updated_at"))
+            if current_updated_at is None or (new_updated_at and new_updated_at > current_updated_at):
+                stored[key] = row
+                changed = True
+    if changed:
+        save_json(file_path, stored)
+        print(f"Inserted/updated {len(rows)} rows in {table_name} table.")
+
+
+def incremental_load(endpoint, table_name):
+    """
+    Fetch data from the API endpoint and store it in the local file.
+    This function will only insert new or updated rows based on the watermark.
+    """
+    ensure_dirs()
+
+    last_watermark = get_watermark(table_name)
+    max_wm = last_watermark
+
+    with retry_mechanism() as sess:
+        for page in iter_pages(endpoint, PAGE_SIZE):
+            fresh_data = []
+            for row in page.get("items", []):
+                updated_at = to_dt(row.get("updated_at"))
+                if last_watermark is None or (updated_at and updated_at > last_watermark):
+                    fresh_data.append(row)
+                    if max_wm is None or (updated_at and updated_at > max_wm):
+                        max_wm = updated_at # to ensure we always keep the actual max upadted_at
+            if fresh_data:
+                bulk_insert(table_name, fresh_data)
+            else:
+                print(f"No new data found for {table_name} in this page.")
+    
+    set_watermark(table_name, max_wm) # updating the max updated_at in the watermark file.
+
+def main():
+    ensure_dirs()
+    for table_name in TABLES:
+        incremental_load(f"/{table_name}", table_name)
+
+if __name__ == "__main__":
+    main()
+
diff --git a/docker-compose.yml b/docker-compose.yml
new file mode 100644
index 0000000..94d9c58
--- /dev/null
+++ b/docker-compose.yml
@@ -0,0 +1,24 @@
+version: "3.8"
+
+services:
+  airflow:
+    image: apache/airflow:2.9.2
+    container_name: moovitamix_airflow
+    environment:
+      AIRFLOW__CORE__LOAD_EXAMPLES: "False"
+      AIRFLOW__WEBSERVER__AUTH_MANAGER: airflow.www.security.NoAuthManager  # no login screen
+      PYTHONPATH: /opt/airflow/repo
+      BASE_URL: "http://host.docker.internal:8000"  
+    volumes:
+      - ./orchestrator/dags:/opt/airflow/dags
+      - ./orchestrator/airflow_home/webserver_config.py:/opt/airflow/webserver_config.py
+      - ./data_ingestion:/opt/airflow/repo/data_ingestion
+      - ./data_store:/opt/airflow/repo/data_store
+      - ./orchestrator/requirements-airflow.txt:/requirements-airflow.txt
+    ports:
+      - "8080:8080"
+    command: >
+      bash -eu -c "
+        pip install --no-cache-dir -r /requirements-airflow.txt &&
+        airflow standalone
+      "
\ No newline at end of file
diff --git a/docs/ANSWERS.md b/docs/ANSWERS.md
index d038faa..fa3123b 100644
--- a/docs/ANSWERS.md
+++ b/docs/ANSWERS.md
@@ -2,22 +2,307 @@
 
 ## _Utilisation de la solution (étape 1 à 3)_
 
-_Inscrire la documentation technique_
+### Moovitamix - Ingestion de Données
+
+Ce projet fournit :
+- une **API FastAPI** qui génère des données factices (tracks, users, listen_history),
+- un flux d’**ingestion incrémentale** (stockage local au format JSON, watermark),
+- une **orchestration quotidienne** via **Airflow** (Docker),
+- quelques **tests unitaires** essentiels. Ceux-ci seront invokés à chaque push via une GitHub action.
+
+---
+
+### 1. Installation
+
+Créez et activez un environnement virtuel, puis installez les dépendances listées dans `requirements.txt`.
+
+```bash
+# Créer un environnement virtuel (exemple avec venv)
+python -m venv .venv
+source .venv/bin/activate   # Sur Windows : .venv\Scripts\activate
+
+# Installer les dépendances
+pip install -r requirements.txt
+
+```
+### 2. Lancer le serveur FastAPI
+
+Déplacez-vous dans le dossier de l’application FastAPI et démarrez le serveur :
+
+```bash
+cd src/moovitamix_fastapi
+python -m uvicorn main:app
+```
+Le serveur sera disponible à l’adresse : http://127.0.0.1:8000
+Documentation Swagger UI : http://127.0.0.1:8000/docs
+
+### 3. Orchestration avec Airflow (Docker)
+
+Prérequis
+
+- Docker & Docker Compose
+
+Démarrage
+
+Depuis la racine du repo (là où se trouve docker-compose.yml) :
+
+```bash
+docker compose up --build
+```
+
+Cela :
+
+- installe les dépendances Airflow (via requirements-airflow.txt),
+- initialise Airflow et lance webserver + scheduler,
+- monte le code d’ingestion dans le conteneur.
+
+Accès à l’UI
+
+Airflow UI : http://localhost:8080
+(configuré en NoAuth pour simplifier la revue)
+
+Déclencher le DAG
+
+Dans l’UI, active le DAG daily_ingest puis clique Play ▶ Trigger DAG.
+
+Le DAG fait :
+
+- incremental_load_tracks
+- incremental_load_users
+- incremental_load_listen_history
+
+Où vont les données ?
+
+Les fichiers JSON sont écrits dans ./data_store/ :
+
+tracks.json, users.json, listen_history.json
+
+watermark.json (pour mémoriser le updated_at max par table)
+
+Important (réseau Docker) : dans docker-compose.yml, l’ingestion utilise BASE_URL=http://host.docker.internal:8000 pour atteindre l’API qui tourne sur ta machine. Si tu changes de port ou d’hôte, adapte cette variable d’environnement.
+
+### 4. Tests untaires (Pytest)
+
+```bash
+pytest -q
+```
+
+Les tests couvrent :
+
+- la pagination (iter_pages),
+- l’ingestion incrémentale (filtre via watermark + mise à jour du watermark),
+- l’upsert local (pas de doublons, remplacement seulement si updated_at plus récent).
+
+### 6. Structure (résumé)
+
+```bash
+.
+├─ data_ingestion/
+│  ├─ fetch_data.py                       # session + pagination
+│  └─ load_data.py                        # upsert local + watermark + incremental_load
+├─ data_store/                            # Fichiers JSON (créé au runtime)
+│ ├─ listen_history.json
+│ ├─ tracks.json
+│ ├─ users.json
+│ └─ watermark.json
+├─ orchestrator/
+│  ├─ dags/
+│  │  └─ daily_ingest.py                  # DAG Airflow (3 tâches d’ingestion)
+│  ├─ airflow_home/                       # Config locale Airflow
+│  ├─ webserver_config.py
+│  └─ requirements-airflow.txt             
+├─ src/moovitamix_fastapi/
+│  ├─ main.py                             # API FastAPI
+│  ├─ classes_out.py
+│  └─ generate_fake_data.py
+├─ test/                                  # tests pytest
+│  ├─ confest.py                          # Fixtures
+│  ├─ test_bulk_insert_and_watermark.py   
+│  ├─ test_classes_out.py
+│  ├─test_incremental_idempotent.py
+│  ├─test_incremental_load.py
+│  └─ test_iter_pages.py
+├─ docker-compose.yml
+├─ pytest.ini
+├─ requirements.txt
+└─ README.md
+```
 
 ## Questions (étapes 4 à 7)
 
 ### Étape 4
 
-_votre réponse ici_
+Il y a trois "layers" au schéma proposé:
+
+A - Raw Layer
+Dans ce niveau, chaque table contient quatre colonnes principales : **ID, UPDATED_AT, INGESTED_AT, PAYLOAD**.  
+Le champ `payload` stocke l’objet JSON brut provenant de l’API.  
+
+```bash
+CREATE TABLE raw_tracks (
+  id           BIGINT PRIMARY KEY,
+  updated_at   TIMESTAMPTZ,
+  ingested_at  TIMESTAMPTZ,
+  payload      JSONB NOT NULL
+);
+```
+
+Pourquoi conserver le payload en JSON ?
+
+1. Évolution du schéma : les APIs changent souvent (ajout/suppression/renommage de champs). Stocker le brut protège contre les breaking changes.
+
+2. Flexibilité : chaque équipe (ML, Analytics, Ops) peut créer ses propres vues transformées à partir de la donnée brute.
+
+
+B - Intermediate layer (Normalization)
+
+Ici, on normalise les données du Raw Layer en tables dimensionnelles et factuelles.
+
+- Les clés du JSON deviennent des colonnes.
+- Les valeurs deviennent des lignes.
+- Pour la table listen_history, le tableau items est explosé : chaque écoute devient une ligne unique.
+- On ajoute une colonne listen_order pour garantir une clé unique (car un utilisateur peut écouter plusieurs fois la même chanson).
+
+```bash
+CREATE TABLE fact_listens (
+  user_id       BIGINT NOT NULL REFERENCES dim_users(user_id),
+  track_id      BIGINT NOT NULL REFERENCES dim_tracks(track_id),
+  listen_order  INT    NOT NULL,               -- position 1..N dans l’historique de l’utilisateur
+  updated_at    TIMESTAMPTZ,                   
+  PRIMARY KEY (user_id, listen_order)
+);
+
+CREATE TABLE dim_users (
+  user_id        BIGINT PRIMARY KEY,
+  first_name     TEXT,
+  last_name      TEXT,
+  email          TEXT,
+  gender         TEXT,
+  favorite_genres TEXT,
+  created_at     TIMESTAMPTZ,
+  updated_at     TIMESTAMPTZ
+);
+
+CREATE TABLE dim_tracks (
+  track_id    BIGINT PRIMARY KEY,
+  name        TEXT NOT NULL,
+  artist      TEXT NOT NULL,
+  songwriters TEXT,
+  duration    TEXT,
+  genres      TEXT,
+  album       TEXT,
+  created_at  TIMESTAMPTZ,
+  updated_at  TIMESTAMPTZ
+);
+```
+
+C - ML ready table (Ce que le modèle pourrait utilisé)
+
+Une table dénormalisée qui rassemble les informations essentielles pour l’entraînement d’un modèle de recommandation :
+
+| user_id | user_name | track_id | track_name | genre | artist  | album   | listen_order | 
+|---------|-----------|----------|------------|-------|---------|---------|--------------|
+| 17081   | Alice     | 87643    | Song A     | Rock  | Band X  | Album 1 | 1            |
+| 17081   | Alice     | 27349    | Song B     | Pop   | Artist Y| Album 2 | 2            |
+| 17081   | Alice     | 41878    | Song C     | Jazz  | Artist Z| Album 3 | 3            |
+
+####Choix du système de base de données
+
+- Court terme : commencer avec PostgreSQL tant que les tables restent sous ~10M de lignes.
+
+   -  Simple d’utilisation.
+   - Supporte bien les PK/FK.
+   - Permet de stocker les payloads bruts en JSON.
+
+- Long terme (100M+ lignes) : PostgreSQL devient limité.
+
+   - Stockage row-based → peu optimal pour de gros scans analytiques.
+   - Parallélisation limitée → pas de MPP natif comme Snowflake, BigQuery ou Redshift.
+   - Scalabilité verticale → tu rajoutes RAM/CPU, mais ça plafonne rapidement.
+
+Pour passer à l’échelle, je recommande un data warehouse distribué comme Snowflake ou Databricks (scalabilité horizontale, MPP, optimisé pour le big data, column-based).
 
 ### Étape 5
 
-_votre réponse ici_
+### Suivi de la santé du pipeline de données
+
+Pour garantir la fiabilité des données ingérées quotidiennement, je mettrais en place un **système de monitoring du pipeline**, basé sur les points suivants :
+
+### Méthode de surveillance
+- **Orchestration** : utilisation d’un orchestrateur (ex. Airflow) avec des logs détaillés pour chaque tâche (ingestion, transformation, stockage).  
+- **Alertes automatisées** : configuration d’alertes (ex. Slack, email) si une tâche échoue, dépasse un temps d’exécution défini, ou charge un volume anormalement faible/élevé de données.  
+- **Historisation des runs** : conserver un registre (table de métadonnées) avec le statut de chaque exécution du pipeline (succès, échec, durée, volume de données).  
+- **Data quality checks** : mise en place de tests automatisés (ex. Great Expectations, dbt tests) pour s’assurer que les données sont complètes et valides.  
+
+### Métriques clés
+- **Disponibilité et statut des tâches**
+  - % de succès/échec par exécution.  
+  - Temps d’exécution moyen vs attendu.  
+
+- **Volumes de données ingérées**
+  - Nombre de chansons, utilisateurs et écoutes ingérés chaque jour.  
+  - Comparaison avec les jours précédents (détection d’anomalies).  
+
+- **Qualité des données**
+  - Champs obligatoires non nuls (ex. `id`, `user_id`, `track_id`).  
+  - Respect des formats (ex. timestamp valide pour `updated_at`).  
+  - Détection de doublons.  
+
+- **Fraîcheur des données**
+  - Vérifier que les données du jour J-1 sont bien arrivées.  
+  - Mesurer le décalage entre l’heure d’ingestion et la dernière mise à jour (*lag*).
 
 ### Étape 6
 
-_votre réponse ici_
+### Automatisation du calcul des recommandations
+
+### Étapes principales
+1. **Ingestion quotidienne** : récupérer les données (users, tracks, historique) via l’API et les stocker.  
+2. **Préparation des données** : nettoyer, normaliser et construire des tables prêtes pour le ML (écoutes, profils utilisateurs, infos des morceaux).  
+3. **Génération des candidats** : pour chaque utilisateur, sélectionner un ensemble de morceaux potentiellement intéressants (basé sur similarité ou popularité).  
+4. **Ranking** : appliquer un modèle (ex. LightGBM ou ALS) qui score les candidats et produit le Top-N morceaux par utilisateur.  
+5. **Publication** : stocker les recommandations dans une table `reco_batch(user_id, track_id, score, generated_at)` et les exposer à l’application (ou via un cache type Redis).  
+
+### Orchestration & Monitoring
+- Utiliser un orchestrateur (Airflow) qui exécute ces étapes chaque nuit.  
+- Surveiller : succès/échec, volumes ingérés, fraîcheur des données.  
+- Déclencher des alertes si une étape échoue ou si le volume est anormal.  
 
 ### Étape 7
 
-_votre réponse ici_
+#### Idée générale
+À chaque arrivée de nouvelles données (écoutes, utilisateurs, morceaux), un **pipeline automatisé** :
+1. Prépare les données,  
+2. Ré-entraîne le modèle,  
+3. Valide les performances,  
+4. Déploie le modèle si les résultats sont satisfaisants.  
+
+---
+
+#### Étapes
+
+1. **Déclenchement**
+   - Tous les jours (cron/Airflow) ou après ingestion complète des données.
+
+2. **Préparation des données**
+   - Lecture des tables (`users`, `tracks`, `listen_history`).  
+   - Filtrage de la période utile (ex. 90 derniers jours).  
+   - Construction des features.  
+   - Sauvegarde du dataset d’entraînement versionné (ex. `training/2025-08-17.parquet`).
+
+3. **Entraînement**
+   - Lancer un script (`train.py`).  
+   - Produire un modèle **candidat** + métriques.
+
+4. **Validation**
+   - Comparer le modèle candidat au modèle **actuel**.  
+   - Si meilleur → continuer.  
+   - Sinon → garder l’ancien modèle et alerter.
+
+5. **Versioning & déploiement**
+   - Sauvegarde du modèle (MLflow ou équivalent).  
+   - Mise à jour du modèle en production (`latest`).
+
+6. **Monitoring**
+   - Suivi des métriques (succès/échec, temps d’exécution, volume de données).  
+   - Alertes (Slack/email) en cas d’anomalie.
\ No newline at end of file
diff --git a/orchestrator/airflow_home/webserver_config.py b/orchestrator/airflow_home/webserver_config.py
new file mode 100644
index 0000000..e5675c6
--- /dev/null
+++ b/orchestrator/airflow_home/webserver_config.py
@@ -0,0 +1,4 @@
+# Disable login entirely: anonymous users get Admin role
+from flask_appbuilder.security.manager import AUTH_DB  # required import
+AUTH_TYPE = AUTH_DB
+AUTH_ROLE_PUBLIC = "Admin"
\ No newline at end of file
diff --git a/orchestrator/dags/daily_ingest.py b/orchestrator/dags/daily_ingest.py
new file mode 100644
index 0000000..d24afa4
--- /dev/null
+++ b/orchestrator/dags/daily_ingest.py
@@ -0,0 +1,37 @@
+from datetime import datetime, timedelta
+from airflow import DAG
+from airflow.operators.python import PythonOperator
+
+from data_ingestion.load_data import incremental_load
+
+default_args = {
+    "owner": "data-eng",
+    "retries": 3,
+    "retry_delay": timedelta(minutes=5),
+    "email_on_failure": False,
+}
+
+with DAG(
+    dag_id="daily_ingest",
+    default_args=default_args,
+    schedule_interval="30 2 * * *",  # Daily at 2:30 AM
+    start_date=datetime(2025, 1, 1),
+    catchup=False,
+    tags=["data-engineering", "music-data-ingestion"],
+) as dag:
+
+    load_tracks = PythonOperator(
+        task_id="incremental_load_tracks",
+        python_callable=incremental_load,
+        op_args=["tracks", "tracks"], # endpoint and table_name
+    )
+    load_users = PythonOperator(
+        task_id="incremental_load_users",
+        python_callable=incremental_load,
+        op_args=["users", "users"],
+    )
+    load_listen_history = PythonOperator(
+        task_id="incremental_load_listen_history",
+        python_callable=incremental_load,
+        op_args=["listen_history", "listen_history"],
+    )
diff --git a/orchestrator/requirements-airflow.txt b/orchestrator/requirements-airflow.txt
new file mode 100644
index 0000000..73395d6
--- /dev/null
+++ b/orchestrator/requirements-airflow.txt
@@ -0,0 +1,3 @@
+apache-airflow==2.9.2
+requests>=2.31
+urllib3>=2.0
\ No newline at end of file
diff --git a/pytest.ini b/pytest.ini
new file mode 100644
index 0000000..665f325
--- /dev/null
+++ b/pytest.ini
@@ -0,0 +1,3 @@
+[pytest]
+pythonpath = .
+testpaths = test
\ No newline at end of file
diff --git a/requirements.txt b/requirements.txt
index 55593fd..d05521a 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -2,4 +2,5 @@ faker
 fastapi
 uvicorn
 fastapi_pagination
-pytest
\ No newline at end of file
+pytest
+requests
\ No newline at end of file
diff --git a/src/moovitamix_fastapi/main.py b/src/moovitamix_fastapi/main.py
index bd1d01f..42261be 100644
--- a/src/moovitamix_fastapi/main.py
+++ b/src/moovitamix_fastapi/main.py
@@ -1,13 +1,17 @@
 from classes_out import ListenHistoryOut, TracksOut, UsersOut
+from typing import TypeVar
 from fastapi import FastAPI, Query
 from fastapi.openapi.docs import get_swagger_ui_html
 from fastapi.responses import RedirectResponse
 from fastapi_pagination import Page, add_pagination, paginate
+from fastapi_pagination.customization import CustomizedPage, UseParamsFields
 from generate_fake_data import FakeDataGenerator
 
-Page = Page.with_custom_options(
-    size=Query(100, ge=1, le=100),
-)
+T = TypeVar("T")
+Page = CustomizedPage[
+    Page[T],
+    UseParamsFields(size=Query(100, ge=1, le=100)),
+]
 
 app = FastAPI(
     title="MooVitamix",
diff --git a/test/conftest.py b/test/conftest.py
new file mode 100644
index 0000000..88e69c7
--- /dev/null
+++ b/test/conftest.py
@@ -0,0 +1,60 @@
+import os
+import json
+import pytest
+import types
+from pathlib import Path
+
+@pytest.fixture(autouse=True)
+def setup_environment(tmp_path, monkeypatch):
+    # Create a temporary data_store directory
+    # Set the BASE_URL environment variable for testing. Fake URL.
+    monkeypatch.setenv("BASE_URL", "http://testserver:8000") 
+
+    from data_ingestion import load_data
+
+    # Create temporary data_store directory
+    test_data_dir = tmp_path / "data_store"
+    test_data_dir.mkdir(parents=True, exist_ok=True)    
+
+
+    monkeypatch.setattr(load_data, "DATA_DIR", test_data_dir, raising=False)
+    monkeypatch.setattr(load_data, "WATERMARK_FILE", test_data_dir / "watermark.json", raising=False)
+
+    test_tables = {}
+    for name, info in load_data.TABLES.items():
+        test_tables[name] = {
+            "pk": info["pk"],
+            "file": test_data_dir / f"{name}.json"
+        }
+    monkeypatch.setattr(load_data, "TABLES", test_tables, raising=False)
+
+    yield
+
+@pytest.fixture
+def fake_session(monkeypatch):
+    class FakeResp:
+        def __init__(self, payload):
+            self._payload = payload
+            self.status_code = 200
+        def raise_for_status(self): pass
+        def json(self): return self._payload
+
+    class FakeSession:
+        def __init__(self): self.get = None
+        def __enter__(self): return self
+        def __exit__(self, *a): pass
+
+    fs = FakeSession()
+    fs.FakeResp = FakeResp
+
+    # Import once here so we can patch
+    import data_ingestion.fetch_data as fetch
+    monkeypatch.setattr(fetch, "retry_mechanism", lambda *a, **k: fs)
+
+    return fs
+
+@pytest.fixture
+def modules():
+    # Handy re-imports
+    from data_ingestion import fetch_data, load_data
+    return types.SimpleNamespace(fetch=fetch_data, load=load_data)
\ No newline at end of file
diff --git a/test/test_bulk_insert_and_watermark.py b/test/test_bulk_insert_and_watermark.py
new file mode 100644
index 0000000..014b54c
--- /dev/null
+++ b/test/test_bulk_insert_and_watermark.py
@@ -0,0 +1,31 @@
+import json
+
+def test_bulk_insert_upsert_updates_only_if_newer(modules, tmp_path):
+    load = modules.load
+
+    # Insert initial row
+    rows1 = [{"id": 10, "name": "A", "updated_at": "2025-08-01T10:00:00"}]
+    load.bulk_insert("tracks", rows1)
+
+    # Older update -> should be ignored
+    rows2 = [{"id": 10, "name": "A-older", "updated_at": "2025-08-01T09:59:59"}]
+    load.bulk_insert("tracks", rows2)
+
+    # Newer update -> should overwrite
+    rows3 = [{"id": 10, "name": "A-new", "updated_at": "2025-08-01T11:00:00"}]
+    load.bulk_insert("tracks", rows3)
+
+    # Read file
+    path = load.TABLES["tracks"]["file"]
+    data = json.loads(path.read_text(encoding="utf-8"))
+    assert "10" in data
+    assert data["10"]["name"] == "A-new"
+
+def test_watermark_roundtrip(modules):
+    load = modules.load
+    assert load.get_watermark("tracks") is None
+    from datetime import datetime
+    ts = datetime.fromisoformat("2025-08-01T11:00:00")
+    load.set_watermark("tracks", ts)
+    out = load.get_watermark("tracks")
+    assert out == ts
\ No newline at end of file
diff --git a/test/test_incremental_idempotent.py b/test/test_incremental_idempotent.py
new file mode 100644
index 0000000..35748f3
--- /dev/null
+++ b/test/test_incremental_idempotent.py
@@ -0,0 +1,44 @@
+import json
+from datetime import datetime
+
+def test_incremental_load_is_idempotent_no_duplicates(fake_session, modules):
+    load = modules.load
+
+    # Seed store and watermark
+    seed_ts = "2025-08-01T12:00:00"
+    load.bulk_insert("tracks", [{"id": 42, "name": "seed", "updated_at": seed_ts}])
+    load.set_watermark("tracks", datetime.fromisoformat(seed_ts))
+
+    # Same record again (same ts) then an older one → both ignored
+    def fake_get(url, params=None, timeout=None):
+        page = params["page"]
+        if page == 1:
+            return fake_session.FakeResp({
+                "items": [{"id": 42, "name": "seed", "updated_at": seed_ts}],
+                "pages": 2,
+            })
+        elif page == 2:
+            return fake_session.FakeResp({
+                "items": [{"id": 42, "name": "seed", "updated_at": "2025-08-01T11:00:00"}],
+                "pages": 2,
+            })
+        else:
+            raise AssertionError("Should not request page > 2")
+
+    fake_session.get = fake_get
+
+    # Running once should be enough, but still want to check if the pipeline is re-triggered with the same input the state remains unchanged. 
+    # Reason why running twice here.
+    load.incremental_load("tracks", "tracks")
+    load.incremental_load("tracks", "tracks")
+
+    # Validate: still a single record, unchanged
+    path = load.TABLES["tracks"]["file"]
+    data = json.loads(path.read_text(encoding="utf-8"))
+    assert list(map(int, data.keys())) == [42]
+    assert data["42"]["name"] == "seed"
+    assert data["42"]["updated_at"] == seed_ts
+
+    # Watermark unchanged (no newer data seen)
+    wm = load.get_watermark("tracks")
+    assert wm.isoformat() == seed_ts
\ No newline at end of file
diff --git a/test/test_incremental_load.py b/test/test_incremental_load.py
new file mode 100644
index 0000000..bc5ac06
--- /dev/null
+++ b/test/test_incremental_load.py
@@ -0,0 +1,42 @@
+import json
+from datetime import datetime
+
+def test_incremental_load_filters_by_watermark_and_updates_it(fake_session, modules):
+    fetch, load = modules.fetch, modules.load
+
+    # Seed watermark so we only accept rows with updated_at > 2025-08-01T12:00:00
+    load.set_watermark("tracks", datetime.fromisoformat("2025-08-01T12:00:00"))
+
+    # Fake paginated responses keyed off the requested page number
+    def fake_get(url, params=None, timeout=None):
+        page = params["page"]
+        if page == 1:
+            return fake_session.FakeResp({
+                "items": [
+                    {"id": 1, "name": "old",  "updated_at": "2025-08-01T11:59:59"},  # ignored
+                    {"id": 2, "name": "new1", "updated_at": "2025-08-01T12:00:01"},  # ingested
+                ],
+                "pages": 2,
+            })
+        elif page == 2:
+            return fake_session.FakeResp({
+                "items": [
+                    {"id": 3, "name": "new2", "updated_at": "2025-08-01T12:30:00"},  # ingested
+                ],
+                "pages": 2,
+            })
+        else:
+            raise AssertionError("Should not request page > 2")
+
+    fake_session.get = fake_get
+
+    # Run incremental load
+    load.incremental_load("tracks", "tracks")
+
+    # Verify stored file contains only ids 2 and 3
+    data = json.loads(load.TABLES["tracks"]["file"].read_text(encoding="utf-8"))
+    assert set(map(int, data.keys())) == {2, 3}
+
+    # Verify watermark advanced to the max updated_at
+    wm = load.get_watermark("tracks")
+    assert wm.isoformat() == "2025-08-01T12:30:00"
\ No newline at end of file
diff --git a/test/test_iter_pages.py b/test/test_iter_pages.py
new file mode 100644
index 0000000..fb4d9ca
--- /dev/null
+++ b/test/test_iter_pages.py
@@ -0,0 +1,28 @@
+from data_ingestion import fetch_data as fetch
+
+def test_iter_pages_paginates(fake_session):
+    # match the signature used by iter_pages
+    def fake_get(url, params=None, timeout=None):
+        page = params["page"]
+        if page == 1:
+            return fake_session.FakeResp({
+                "items": [{"id": 1, "updated_at": "2025-08-01T00:00:00"}],
+                "pages": 2
+            })
+        elif page == 2:
+            return fake_session.FakeResp({
+                "items": [{"id": 2, "updated_at": "2025-08-02T00:00:00"}],
+                "pages": 2
+            })
+        else:
+            raise AssertionError("Should not request page > 2")
+
+    fake_session.get = fake_get
+
+    # run
+    pages = list(fetch.iter_pages("tracks", size=100))
+
+    # assert
+    assert len(pages) == 2
+    assert pages[0]["items"][0]["id"] == 1
+    assert pages[1]["items"][0]["id"] == 2
\ No newline at end of file