colin/colibre
1
0
Fork 0
Code Issues 30 Pull Requests Actions Packages Projects Releases 33 Activity

Merge branch 'feature/ui_tests' into dev

Browse Source
This commit is contained in:
Colin Maudry
2026-02-05 15:05:59 +01:00
parent 39af4bafbe 3beedbcfe4
commit 6891aabd4d
10 changed files with 234 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+17
View File
@@ -22,5 +22,22 @@ dependencies = [
[project.optional-dependencies]
dev = [
"pytest",
"pytest-env",
"pre-commit",
"selenium",
"webdriver-manager",
"dash[testing]",
]
[tool.pytest.ini_options]
pythonpath = [
"src"
]
testpaths = [
"tests"
]
env = [
"DATA_FILE_PARQUET_PATH=tests/test.parquet"
]
addopts = "-p no:warnings"
src/app.py
+5 -1
View File
@@ -9,6 +9,10 @@ from flask import Response
load_dotenv()
# if os.getenv("PYTEST_CURRENT_TEST"):
# os.environ["DATA_FILE_PARQUET_PATH"]
development = os.getenv("DEVELOPMENT").lower() == "true"
meta_tags = [
@@ -22,7 +26,7 @@ meta_tags = [
if development:
meta_tags.append({"name": "robots", "content": "noindex"})
app = Dash(
app: Dash = Dash(
title="decp.info",
use_pages=True,
compress=True,
src/figures.py
-1
View File
@@ -330,7 +330,6 @@ class DataTable(dash_table.DataTable):
},
sort_action=sort_action,
sort_mode="multi",
sort_by=[],
row_deletable=False,
page_current=0,
style_cell_conditional=style_cell_conditional,
src/pages/acheteur.py
+43 -56
View File
@@ -1,5 +1,4 @@
import datetime
import uuid
import dash_bootstrap_components as dbc
import polars as pl
@@ -12,7 +11,6 @@ from dash import (
clientside_callback,
dcc,
html,
no_update,
register_page,
)
@@ -58,6 +56,9 @@ datatable = html.Div(
className="marches_table",
children=DataTable(
dtid="acheteur_datatable",
persistence=True,
persistence_type="local",
persisted_props=["filter_query", "sort_by"],
page_action="custom",
filter_action="custom",
sort_action="custom",
@@ -70,8 +71,6 @@ datatable = html.Div(
layout = [
dcc.Store(id="acheteur_data", storage_type="memory"),
dcc.Store(id="acheteur-hidden-columns", storage_type="local"),
dcc.Store(id="acheteur-filters", storage_type="local"),
dcc.Store(id="acheteur-sort", storage_type="local"),
dcc.Store(id="filter-cleanup-trigger-acheteur"),
dcc.Location(id="acheteur_url", refresh="callback-nav"),
html.Div(
@@ -218,26 +217,40 @@ layout = [
)
def update_acheteur_infos(url):
acheteur_siret = url.split("/")[-1]
if len(acheteur_siret) != 14:
acheteur_siret = (
f"Le SIRET renseigné doit faire 14 caractères ({acheteur_siret})"
)
# if len(acheteur_siret) != 14:
# acheteur_siret = (
# f"Le SIRET renseigné doit faire 14 caractères ({acheteur_siret})"
# )
data = get_annuaire_data(acheteur_siret)
data_etablissement = data["matching_etablissements"][0]
acheteur_map = point_on_map(
data_etablissement["latitude"], data_etablissement["longitude"]
)
code_departement, nom_departement, nom_region = get_departement_region(
data_etablissement["code_postal"]
)
departement = f"{nom_departement} ({code_departement})"
lien_annuaire = (
f"https://annuaire-entreprises.data.gouv.fr/etablissement/{acheteur_siret}"
)
data_etablissement = data.get("matching_etablissements") if data else None
if data_etablissement:
data_etablissement = data_etablissement[0]
acheteur_map = point_on_map(
data_etablissement["latitude"], data_etablissement["longitude"]
)
code_departement, nom_departement, nom_region = get_departement_region(
data_etablissement["code_postal"]
)
departement = f"{nom_departement} ({code_departement})"
lien_annuaire = (
f"https://annuaire-entreprises.data.gouv.fr/etablissement/{acheteur_siret}"
)
raison_sociale = data["nom_raison_sociale"]
libelle_commune = data_etablissement["libelle_commune"]
else:
acheteur_map = html.Div()
code_departement, nom_departement, nom_region = "", "", ""
departement = ""
lien_annuaire = ""
raison_sociale = ""
libelle_commune = ""
return (
acheteur_siret,
data["nom_raison_sociale"],
data_etablissement["libelle_commune"],
raison_sociale,
libelle_commune,
acheteur_map,
departement,
nom_region,
@@ -306,16 +319,17 @@ def get_acheteur_marches_data(url, acheteur_year: str) -> tuple:
Output("btn-download-filtered-data-acheteur", "children"),
Output("btn-download-filtered-data-acheteur", "title"),
Output("filter-cleanup-trigger-acheteur", "data", allow_duplicate=True),
Input("acheteur_url", "href"),
Input("acheteur_data", "data"),
Input("acheteur_datatable", "page_current"),
Input("acheteur_datatable", "page_size"),
Input("acheteur-filters", "data"),
Input("acheteur-sort", "data"),
Input("acheteur_datatable", "filter_query"),
Input("acheteur_datatable", "sort_by"),
State("acheteur_datatable", "data_timestamp"),
prevent_initial_call=True,
)
def get_last_marches_data(
data, page_current, page_size, filter_query, sort_by, data_timestamp
href, data, page_current, page_size, filter_query, sort_by, data_timestamp
) -> tuple:
return prepare_table_data(
data, data_timestamp, filter_query, page_current, page_size, sort_by, "acheteur"
@@ -360,8 +374,8 @@ def download_acheteur_data(
State("acheteur_data", "data"),
Input("btn-download-filtered-data-acheteur", "n_clicks"),
State("acheteur_nom", "children"),
State("acheteur-filters", "data"),
State("acheteur-sort", "data"),
State("acheteur_datatable", "filter_query"),
State("acheteur_datatable", "sort_by"),
State("acheteur_datatable", "hidden_columns"),
prevent_initial_call=True,
)
@@ -406,19 +420,16 @@ clientside_callback(
@callback(
Output("acheteur-hidden-columns", "data", allow_duplicate=True),
Output("filter-cleanup-trigger-acheteur", "data", allow_duplicate=True),
Input("acheteur_column_list", "selected_rows"),
State("acheteur-filters", "data"),
prevent_initial_call=True,
)
def update_hidden_columns_from_checkboxes(selected_columns, filter_query):
trigger_cleanup = str(uuid.uuid4()) if filter_query else no_update
def update_hidden_columns_from_checkboxes(selected_columns):
if selected_columns:
selected_columns = [columns[i] for i in selected_columns]
hidden_columns = [col for col in columns if col not in selected_columns]
return hidden_columns, trigger_cleanup
return hidden_columns
else:
return [], trigger_cleanup
return []
@callback(
@@ -458,30 +469,6 @@ def toggle_acheteur_columns(click_open, click_close, is_open):
return is_open
@callback(
Output("acheteur-filters", "data"), Input("acheteur_datatable", "filter_query")
)
def sync_filters_to_local_storage(filter_query):
return filter_query
@callback(Output("acheteur-sort", "data"), Input("acheteur_datatable", "sort_by"))
def sync_sort_to_local_storage(sort_by):
return sort_by
@callback(
Output("acheteur_datatable", "filter_query", allow_duplicate=True),
Output("acheteur_datatable", "sort_by", allow_duplicate=True),
Input("acheteur_url", "href"),
State("acheteur-filters", "data"),
State("acheteur-sort", "data"),
prevent_initial_call=True,
)
def sync_local_storage_to_datatable(href, filter_query, sort_by):
return filter_query, sort_by
@callback(
Output("acheteur_datatable", "filter_query", allow_duplicate=True),
Output("acheteur_datatable", "sort_by"),
src/pages/tableau.py
+9 -17
View File
@@ -55,6 +55,9 @@ datatable = html.Div(
className="marches_table",
children=DataTable(
dtid="tableau_datatable",
persisted_props=["filter_query", "sort_by"],
persistence_type="local",
persistence=True,
page_size=20,
page_action="custom",
filter_action="custom",
@@ -68,8 +71,6 @@ layout = [
dcc.Location(id="tableau_url", refresh=False),
dcc.Store(id="filter-cleanup-trigger-tableau"),
dcc.Store(id="tableau-hidden-columns", storage_type="local"),
dcc.Store(id="tableau-filters", storage_type="local"),
dcc.Store(id="tableau-sort", storage_type="local"),
dcc.Store(id="tableau-table"),
html.Script(
type="application/ld+json",
@@ -285,14 +286,15 @@ layout = [
Output("btn-download-data", "children"),
Output("btn-download-data", "title"),
Output("filter-cleanup-trigger-tableau", "data", allow_duplicate=True),
Input("tableau_url", "href"),
Input("tableau_datatable", "page_current"),
Input("tableau_datatable", "page_size"),
Input("tableau-filters", "data"),
Input("tableau-sort", "data"),
Input("tableau_datatable", "filter_query"),
Input("tableau_datatable", "sort_by"),
State("tableau_datatable", "data_timestamp"),
prevent_initial_call=True,
)
def update_table(page_current, page_size, filter_query, sort_by, data_timestamp):
def update_table(href, page_current, page_size, filter_query, sort_by, data_timestamp):
# if ctx.triggered_id != "url":
# search_params = None
# else:
@@ -337,8 +339,8 @@ def download_data(n_clicks, filter_query, sort_by, hidden_columns: list = None):
Output("tableau_url", "search"),
Output("filter-cleanup-trigger-tableau", "data"),
Input("tableau_url", "search"),
State("tableau-filters", "data"),
State("tableau-sort", "data"),
State("tableau_datatable", "filter_query"),
State("tableau_datatable", "sort_by"),
)
def restore_view_from_url(search, stored_filters, stored_sort):
if not search and not stored_filters:
@@ -512,16 +514,6 @@ def toggle_tableau_columns(click_open, click_close, is_open):
return is_open
@callback(Output("tableau-filters", "data"), Input("tableau_datatable", "filter_query"))
def sync_filters_to_local_storage(filter_query):
return filter_query
@callback(Output("tableau-sort", "data"), Input("tableau_datatable", "sort_by"))
def sync_sort_to_local_storage(sort_by):
return sort_by
@callback(
Output("tableau_datatable", "filter_query", allow_duplicate=True),
Output("tableau_datatable", "sort_by", allow_duplicate=True),
src/pages/titulaire.py
+38 -48
View File
@@ -56,6 +56,9 @@ datatable = html.Div(
className="marches_table",
children=DataTable(
dtid="titulaire_datatable",
persistence=True,
persistence_type="local",
persisted_props=["filter_query", "sort_by"],
page_action="custom",
filter_action="custom",
sort_action="custom",
@@ -68,8 +71,6 @@ datatable = html.Div(
layout = [
dcc.Store(id="titulaire_data", storage_type="memory"),
dcc.Store(id="titulaire-hidden-columns", storage_type="local"),
dcc.Store(id="titulaire-filters", storage_type="local"),
dcc.Store(id="titulaire-sort", storage_type="local"),
dcc.Store(id="filter-cleanup-trigger-titulaire"),
dcc.Location(id="titulaire_url", refresh="callback-nav"),
html.Div(
@@ -217,26 +218,38 @@ layout = [
)
def update_titulaire_infos(url):
titulaire_siret = url.split("/")[-1]
if len(titulaire_siret) != 14:
titulaire_siret = (
f"Le SIRET renseigné doit faire 14 caractères ({titulaire_siret})"
)
data = get_annuaire_data(titulaire_siret)
data_etablissement = data["matching_etablissements"][0]
titulaire_map = point_on_map(
data_etablissement["latitude"], data_etablissement["longitude"]
)
code_departement, nom_departement, nom_region = get_departement_region(
data_etablissement["code_postal"]
)
departement = f"{nom_departement} ({code_departement})"
lien_annuaire = (
f"https://annuaire-entreprises.data.gouv.fr/etablissement/{titulaire_siret}"
)
data_etablissement = data.get("matching_etablissements") if data else None
if data_etablissement:
data_etablissement = data_etablissement[0]
titulaire_map = point_on_map(
data_etablissement["latitude"], data_etablissement["longitude"]
)
code_departement, nom_departement, nom_region = get_departement_region(
data_etablissement["code_postal"]
)
departement = f"{nom_departement} ({code_departement})"
lien_annuaire = (
f"https://annuaire-entreprises.data.gouv.fr/etablissement/{titulaire_siret}"
)
raison_sociale = data["nom_raison_sociale"]
libelle_commune = data_etablissement["libelle_commune"]
else:
titulaire_map = html.Div()
code_departement, nom_departement, nom_region = "", "", ""
departement = ""
lien_annuaire = ""
raison_sociale = html.Span(
f"N° SIREN inconnu de l'INSEE ({titulaire_siret[:9]})"
)
libelle_commune = ""
return (
titulaire_siret,
data["nom_raison_sociale"],
data_etablissement["libelle_commune"],
raison_sociale,
libelle_commune,
titulaire_map,
departement,
nom_region,
@@ -314,16 +327,17 @@ def get_titulaire_marches_data(url, titulaire_year: str) -> tuple:
Output("btn-download-filtered-data-titulaire", "children"),
Output("btn-download-filtered-data-titulaire", "title"),
Output("filter-cleanup-trigger-titulaire", "data", allow_duplicate=True),
Input(component_id="titulaire_url", component_property="href"),
Input("titulaire_data", "data"),
Input("titulaire_datatable", "page_current"),
Input("titulaire_datatable", "page_size"),
Input("titulaire-filters", "data"),
Input("titulaire-sort", "data"),
Input("titulaire_datatable", "filter_query"),
Input("titulaire_datatable", "sort_by"),
State("titulaire_datatable", "data_timestamp"),
config_prevent_initial_callbacks=True,
)
def get_last_marches_data(
data, page_current, page_size, filter_query, sort_by, data_timestamp
href, data, page_current, page_size, filter_query, sort_by, data_timestamp
) -> list[dict]:
return prepare_table_data(
data,
@@ -374,8 +388,8 @@ def download_titulaire_data(
State("titulaire_data", "data"),
Input("btn-download-filtered-data-titulaire", "n_clicks"),
State("titulaire_nom", "children"),
State("titulaire-filters", "data"),
State("titulaire-sort", "data"),
State("titulaire_datatable", "filter_query"),
State("titulaire_datatable", "sort_by"),
State("titulaire_datatable", "hidden_columns"),
prevent_initial_call=True,
)
@@ -469,30 +483,6 @@ def toggle_titulaire_columns(click_open, click_close, is_open):
return is_open
@callback(
Output("titulaire-filters", "data"), Input("titulaire_datatable", "filter_query")
)
def sync_filters_to_local_storage(filter_query):
return filter_query
@callback(Output("titulaire-sort", "data"), Input("titulaire_datatable", "sort_by"))
def sync_sort_to_local_storage(sort_by):
return sort_by
@callback(
Output("titulaire_datatable", "filter_query", allow_duplicate=True),
Output("titulaire_datatable", "sort_by", allow_duplicate=True),
Input("titulaire_url", "href"),
State("titulaire-filters", "data"),
State("titulaire-sort", "data"),
prevent_initial_call=True,
)
def sync_local_storage_to_datatable(href, filter_query, sort_by):
return filter_query, sort_by
@callback(
Output("titulaire_datatable", "filter_query", allow_duplicate=True),
Output("titulaire_datatable", "sort_by"),
src/utils.py
+9 -4
View File
@@ -9,7 +9,7 @@ import dash
import polars as pl
import polars.selectors as cs
from dash import no_update
from httpx import get, post
from httpx import HTTPError, get, post
from polars.exceptions import ComputeError
from unidecode import unidecode
@@ -208,8 +208,13 @@ def format_values(dff: pl.DataFrame) -> pl.DataFrame:
def get_annuaire_data(siret: str) -> dict:
url = f"https://recherche-entreprises.api.gouv.fr/search?q={siret}"
response = get(url)
return response.json()["results"][0]
try:
response = get(url).raise_for_status()
response = response.json()["results"][0]
except (HTTPError, IndexError):
response = None
logger.warning("Could not fetch data from recherche-entreprises.api.")
return response
def get_decp_data() -> pl.DataFrame:
@@ -601,7 +606,7 @@ def prepare_table_data(
trigger_cleanup = no_update if source_table == "tableau" else str(uuid.uuid4())
# Application des tris
if len(sort_by) > 0:
if sort_by and len(sort_by) > 0:
lff = sort_table_data(lff, sort_by)
# Matérialisation des filtres
tests/__init__.py
View File
tests/conftest.py
+50
View File
@@ -0,0 +1,50 @@
import datetime
import os
import polars as pl
import pytest
from selenium.webdriver.chrome.options import Options
@pytest.fixture(scope="session", autouse=True)
def test_data():
data = [
{
"uid": "1",
"id": "1",
"acheteur_nom": "ACHETEUR 1",
"acheteur_id": "a1",
"titulaire_nom": "TITULAIRE 1",
"titulaire_id": "t1",
"montant": 10,
"dateNotification": datetime.date(2025, 1, 1),
"codeCPV": "71600000",
"donneesActuelles": True,
"acheteur_departement_code": "75",
"acheteur_departement_nom": "Paris",
"acheteur_commune_nom": "Paris",
"titulaire_departement_code": "35",
"titulaire_departement_nom": "Ille-et-Vilaine",
"titulaire_commune_nom": "Rennes",
"titulaire_distance": 10,
"titulaire_typeIdentifiant": "SIRET",
"objet": "Objet test",
"dureeRestanteMois": 12,
"lieuExecution_code": "75001",
"sourceFile": "test.xml",
"sourceDataset": "test_dataset",
"datePublicationDonnees": datetime.date(2025, 1, 1),
}
]
path = "tests/test.parquet"
path = os.path.abspath(path)
print(f"Writing test data to: {path}") # <-- This will show you the real path
pl.DataFrame(data).write_parquet("tests/test.parquet")
yield path
def pytest_setup_options():
options = Options()
options.add_argument("--window-size=1200,800")
return options
tests/test_main.py
+63
View File
@@ -0,0 +1,63 @@
from time import sleep
from dash.testing.composite import DashComposite
from selenium.webdriver import Keys
from selenium.webdriver.common.by import By
from selenium.webdriver.remote.webelement import WebElement
def test_001_logo_and_search(dash_duo: DashComposite):
from src.app import app
dash_duo.start_server(app)
dash_duo.wait_for_text_to_equal(".logo > h1", "decp.info", timeout=4)
assert dash_duo.find_element(".logo > h1").text == "decp.info"
for org_type in ["acheteur", "titulaire"]:
name = f"{org_type.upper()} 1"
search_bar: WebElement = dash_duo.find_element("#search")
dash_duo.clear_input(search_bar)
search_bar.send_keys(name)
search_bar.send_keys(Keys.ENTER)
dash_duo.wait_for_element(f"#results_{org_type}_datatable", timeout=2)
result_table: WebElement = dash_duo.find_element(
f"#results_{org_type}_datatable tbody"
)
assert len(result_table.find_elements(by=By.TAG_NAME, value="tr")) == 2, (
"The search should return only one result"
) # header row + 1 result
assert (
result_table.find_element(
by=By.CSS_SELECTOR, value=f'td[data-dash-column="{org_type}_nom"]'
).text
== name
), f"The search result should have the right {org_type} name"
def test_002_filter_persistence(dash_duo: DashComposite):
from src.app import app
dash_duo.start_server(app)
dash_duo.wait_for_text_to_equal(".logo > h1", "decp.info", timeout=4)
def open_page_and_check_filter_input():
dash_duo.wait_for_page(f"{dash_duo.server_url}/{page}")
filter_input_selector = (
'.marches_table th[data-dash-column="uid"] input[type="text"]'
)
dash_duo.wait_for_element(filter_input_selector, timeout=2)
_filter_input: WebElement = dash_duo.find_element(filter_input_selector)
return _filter_input
for page in ["tableau", "acheteurs/a1", "titulaires/t1"]:
print("page:", page)
filter_input = open_page_and_check_filter_input()
filter_input.send_keys("11") # a UID that doesn't exist
filter_input.send_keys(Keys.ENTER)
sleep(1)
filter_input = open_page_and_check_filter_input()
assert filter_input.get_attribute("value") == "11"