location test, OpenAPI

.github/workflows/build-publish.yml

@@ -4,8 +4,9 @@ on:
   release:
     types:
       - published
-  pull_request:
-    branches: [ "main" ]
+  push:
+    tags:
+      - 'preview-**'
 
 jobs:
   build-publish:

requirements-dev.txt

@@ -2,3 +2,5 @@ pytest~=9.0.1
 pytest-cov~=7.0.0
 requests-mock~=1.12.1
 ruff~=0.14.5
+# openapi-core 0.19.5 conflicts with werkzeug 3.1.3
+openapi-core~=0.19.4

tests/integration/conftest.py

@@ -17,26 +17,30 @@
 """
 
 import os
-from typing import Generator
+from typing import Callable, Generator
 
 import pytest
 
+from tests.integration.helpers import OpenApiFlaskClient, TestApp, empty_all_tables
 from webapp import launch
 from webapp.common.flask_app import App
 from webapp.common.sqlalchemy import SQLAlchemy
 from webapp.extensions import db as flask_sqlalchemy
 
 
-@pytest.fixture
+@pytest.fixture(scope='session')
 def flask_app() -> Generator[App, None, None]:
     # Load default development config instead of config.yaml for testing to avoid issues with local setups
     os.environ['CONFIG_FILE'] = os.environ.get('TEST_CONFIG_FILE', 'config_dist_dev.yaml')
 
     app = launch(
+        app_class=TestApp,
         config_overrides={
             'TESTING': True,
             'DEBUG': True,
-        }
+            'DISABLE_EVENTS': True,
+            'SERVER_NAME': 'http://localhost:5010',
+        },
     )
 
     with app.app_context():
@@ -46,10 +50,32 @@ def flask_app() -> Generator[App, None, None]:
         yield app  # type: ignore
 
 
-@pytest.fixture
+@pytest.fixture(scope='function')
 def db(flask_app: App) -> Generator[SQLAlchemy, None, None]:
     """
     Yields the database as a function-scoped fixture with freshly emptied tables.
     """
 
+    empty_all_tables(db=flask_sqlalchemy)
+
     yield flask_sqlalchemy
+
+
+@pytest.fixture
+def test_cli(flask_app: TestApp) -> Generator[Callable, None, None]:
+    def check_cli(fnc: Callable, *args, **kwargs) -> None:
+        runner = flask_app.test_cli_runner()
+        cli_result = runner.invoke(fnc, args=args, **kwargs)
+
+        if cli_result.exception:
+            raise cli_result.exception
+
+        assert cli_result.exit_code == 0
+
+    yield check_cli
+
+
+@pytest.fixture
+def test_client(flask_app: TestApp) -> Generator[OpenApiFlaskClient, None, None]:
+    with flask_app.test_client() as client:
+        yield client  # type: ignore

tests/integration/helpers.py

@@ -0,0 +1,87 @@
+"""
+Open ChargePoint DataBase OCPDB
+Copyright (C) 2025 binary butterfly GmbH
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program.  If not, see <http://www.gnu.org/licenses/>.
+"""
+
+import os
+from typing import Any
+
+from flask.testing import FlaskClient
+from flask_openapi.generator import generate_openapi
+from openapi_core import OpenAPI
+from openapi_core.contrib.werkzeug import WerkzeugOpenAPIRequest, WerkzeugOpenAPIResponse
+from sqlalchemy import text
+from werkzeug.test import TestResponse
+
+from webapp.common.flask_app import App
+from webapp.common.sqlalchemy import SQLAlchemy
+
+OPENAPI_BY_REALM = {}
+
+
+class OpenApiFlaskClient(FlaskClient):
+    openapi_realm: str | None = None
+
+    def __init__(self, *args: Any, openapi_realm: str | None = None, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+        self.openapi_realm = openapi_realm
+
+    def open(self, *args: Any, **kwargs: Any) -> TestResponse:
+        response = super().open(*args, **kwargs)
+
+        if self.openapi_realm is None or os.environ.get('OPENAPI_SKIP_VALIDATION'):
+            return response
+
+        if self.openapi_realm not in OPENAPI_BY_REALM:
+            openapi_dict = generate_openapi(self.openapi_realm)
+            openapi_dict = self.no_additional_properties(openapi_dict)
+
+            OPENAPI_BY_REALM[self.openapi_realm] = OpenAPI.from_dict(openapi_dict)
+
+        OPENAPI_BY_REALM[self.openapi_realm].validate_response(
+            WerkzeugOpenAPIRequest(response.request),
+            WerkzeugOpenAPIResponse(response),
+        )
+
+        return response
+
+    def no_additional_properties(self, data: Any):
+        if isinstance(data, dict):
+            # Patch in additionalProperties in case of an object field if it's not already set
+            if data.get('type') == 'object' and 'additionalProperties' not in data:
+                data['additionalProperties'] = False
+
+            return {key: self.no_additional_properties(value) for key, value in data.items()}
+
+        if isinstance(data, list):
+            return [self.no_additional_properties(item) for item in data]
+
+        return data
+
+
+class TestApp(App):
+    test_client_class = OpenApiFlaskClient
+
+
+def empty_all_tables(db: SQLAlchemy) -> None:
+    """
+    empty all tables in the database
+    (this is much faster than completely deleting the database and creating a new one)
+    """
+    db.session.close()
+    with db.engine.connect() as connection:
+        connection.execute(text(f'TRUNCATE {", ".join(db.metadata.tables.keys())} RESTART IDENTITY;'))
+        connection.commit()

tests/integration/model_generators/business.py

@@ -0,0 +1,38 @@
+"""
+Open ChargePoint DataBase OCPDB
+Copyright (C) 2025 binary butterfly GmbH
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program.  If not, see <http://www.gnu.org/licenses/>.
+"""
+
+from webapp.models import Business
+
+BUSINESS_1_NAME = 'Electro Inc'
+BUSINESS_2_NAME = 'Power Inc'
+
+
+def get_business(**kwargs) -> Business:
+    default_data = {
+        'name': BUSINESS_1_NAME,
+    }
+    data = {**default_data, **kwargs}
+    return Business(**data)
+
+
+def get_business_1(**kwargs) -> Business:
+    return get_business(**kwargs)
+
+
+def get_business_2(**kwargs) -> Business:
+    return get_business(name=BUSINESS_2_NAME, **kwargs)

tests/integration/model_generators/connector.py

@@ -0,0 +1,56 @@
+"""
+Open ChargePoint DataBase OCPDB
+Copyright (C) 2025 binary butterfly GmbH
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program.  If not, see <http://www.gnu.org/licenses/>.
+"""
+
+from datetime import datetime, timezone
+
+from webapp.models import Connector
+from webapp.models.connector import ConnectorFormat, ConnectorType, PowerType
+
+CONNECTOR_1_UID = 'CONNECTOR-1'
+CONNECTOR_2_UID = 'CONNECTOR-2'
+
+
+def get_connector(**kwargs) -> Connector:
+    default_data = {
+        'uid': CONNECTOR_1_UID,
+        'standard': ConnectorType.IEC_62196_T2,
+        'format': ConnectorFormat.SOCKET,
+        'power_type': PowerType.AC_3_PHASE,
+        'max_voltage': 400,
+        'max_amperage': 32,
+        'max_electric_power': 22000,
+        'last_updated': datetime.now(tz=timezone.utc),
+    }
+    data = {**default_data, **kwargs}
+    return Connector(**data)
+
+
+def get_ac_connector(**kwargs) -> Connector:
+    return get_connector(**kwargs)
+
+
+def get_dc_connector(**kwargs) -> Connector:
+    return get_connector(
+        power_type=PowerType.DC,
+        standard=ConnectorType.IEC_62196_T2_COMBO,
+        format=ConnectorFormat.CABLE,
+        max_voltage=400,
+        max_amperage=350,
+        max_electric_power=150000,
+        **kwargs,
+    )

tests/integration/model_generators/evse.py

@@ -0,0 +1,107 @@
+"""
+Open ChargePoint DataBase OCPDB
+Copyright (C) 2025 binary butterfly GmbH
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program.  If not, see <http://www.gnu.org/licenses/>.
+"""
+
+from datetime import datetime, timezone
+
+from tests.integration.model_generators.connector import get_ac_connector, get_dc_connector
+from webapp.models import Evse
+from webapp.models.evse import EvseStatus
+
+EVSE_UID_1 = 'EVSE-1'
+EVSE_UID_2 = 'EVSE-2'
+EVSE_UID_3 = 'EVSE-3'
+EVSE_UID_4 = 'EVSE-4'
+EVSE_UID_5 = 'EVSE-5'
+EVSE_UID_6 = 'EVSE-6'
+
+
+def get_evse(**kwargs) -> Evse:
+    default_data = {
+        'uid': EVSE_UID_1,
+        'evse_id': EVSE_UID_1,
+        'status': EvseStatus.AVAILABLE,
+        'last_updated': datetime.now(timezone.utc),
+    }
+    data = {**default_data, **kwargs}
+    return Evse(**data)
+
+
+def get_evse_1(**kwargs) -> Evse:
+    return get_evse(**kwargs)
+
+
+def get_evse_2(**kwargs) -> Evse:
+    return get_evse(uid=EVSE_UID_2, **kwargs)
+
+
+def get_evse_3(**kwargs) -> Evse:
+    return get_evse(uid=EVSE_UID_3, **kwargs)
+
+
+def get_evse_4(**kwargs) -> Evse:
+    return get_evse(uid=EVSE_UID_4, **kwargs)
+
+
+def get_evse_5(**kwargs) -> Evse:
+    return get_evse(uid=EVSE_UID_5, **kwargs)
+
+
+def get_evse_6(**kwargs) -> Evse:
+    return get_evse(uid=EVSE_UID_6, **kwargs)
+
+
+def get_full_evse_1(**kwargs) -> Evse:
+    return get_evse_1(
+        connectors=[get_ac_connector()],
+        **kwargs,
+    )
+
+
+def get_full_evse_2(**kwargs) -> Evse:
+    return get_evse_2(
+        connectors=[get_ac_connector()],
+        **kwargs,
+    )
+
+
+def get_full_evse_3(**kwargs) -> Evse:
+    return get_evse_3(
+        connectors=[get_dc_connector()],
+        **kwargs,
+    )
+
+
+def get_full_evse_4(**kwargs) -> Evse:
+    return get_evse_4(
+        connectors=[get_dc_connector()],
+        **kwargs,
+    )
+
+
+def get_full_evse_5(**kwargs) -> Evse:
+    return get_evse_5(
+        connectors=[get_ac_connector()],
+        **kwargs,
+    )
+
+
+def get_full_evse_6(**kwargs) -> Evse:
+    return get_evse_6(
+        connectors=[get_ac_connector()],
+        **kwargs,
+    )