From 3ba93598811624c404c6e7edb487a0f4471ea9c1 Mon Sep 17 00:00:00 2001
From: AlexeerCT <acazares@aduanasoft.com.mx>
Date: Tue, 24 Mar 2026 15:20:08 -0500
Subject: [PATCH 01/10] Update CI workflow to include database migration before
 running tests

- Added an environment variable for the test database URL and executed an Alembic migration to ensure the database schema is up-to-date prior to running tests.
- This change aims to improve the reliability of test execution by ensuring the correct database state.
---
 .gitea/workflows/build.yml | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/.gitea/workflows/build.yml b/.gitea/workflows/build.yml
index 7edc0850..34b00be4 100644
--- a/.gitea/workflows/build.yml
+++ b/.gitea/workflows/build.yml
@@ -64,6 +64,8 @@ jobs:
           python -m pip install --upgrade pip
           pip install -r "$GITHUB_WORKSPACE/backend/requirements.txt"
           cd "$GITHUB_WORKSPACE/backend"
+          export DATABASE_URL="$TEST_DATABASE_URL"
+          alembic upgrade head
           pytest -q tests -v -ra -s
 
   build:

From 4b2a3da6113f27f61aa687e4d73bebf6e495fe03 Mon Sep 17 00:00:00 2001
From: AlexeerCT <acazares@aduanasoft.com.mx>
Date: Tue, 24 Mar 2026 15:28:41 -0500
Subject: [PATCH 02/10] Refactor database URL handling in Alembic environment

- Introduced a new function to normalize database URLs for Alembic, ensuring compatibility with different PostgreSQL drivers.
- Updated the `get_database_url` function to prioritize the use of the `TEST_DATABASE_URL` environment variable for CI and testing scenarios.
- Enhanced error messaging to clarify configuration requirements for database connections.

This change aims to improve the flexibility and reliability of database connections in the testing environment.
---
 .gitea/workflows/build.yml |  2 +-
 backend/alembic/env.py     | 25 +++++++++++++++++++++++--
 2 files changed, 24 insertions(+), 3 deletions(-)

diff --git a/.gitea/workflows/build.yml b/.gitea/workflows/build.yml
index 34b00be4..23a2f909 100644
--- a/.gitea/workflows/build.yml
+++ b/.gitea/workflows/build.yml
@@ -64,7 +64,7 @@ jobs:
           python -m pip install --upgrade pip
           pip install -r "$GITHUB_WORKSPACE/backend/requirements.txt"
           cd "$GITHUB_WORKSPACE/backend"
-          export DATABASE_URL="$TEST_DATABASE_URL"
+          export TEST_DATABASE_URL="$TEST_DATABASE_URL"
           alembic upgrade head
           pytest -q tests -v -ra -s
 
diff --git a/backend/alembic/env.py b/backend/alembic/env.py
index fbe67248..bfbd92f1 100644
--- a/backend/alembic/env.py
+++ b/backend/alembic/env.py
@@ -18,9 +18,29 @@ logger = logging.getLogger(__name__)
 config = context.config
 
 
+def _normalize_alembic_sqlalchemy_url(url: str) -> str:
+    """Alembic usa el driver síncrono psycopg2; normaliza DSN típicos de app/tests."""
+    url = url.strip()
+    if url.startswith("postgresql+asyncpg://"):
+        return url.replace("postgresql+asyncpg://", "postgresql+psycopg2://", 1)
+    if url.startswith("postgresql+psycopg2://"):
+        return url
+    if url.startswith("postgresql://"):
+        return url.replace("postgresql://", "postgresql+psycopg2://", 1)
+    if url.startswith("postgres://"):
+        return url.replace("postgres://", "postgresql+psycopg2://", 1)
+    return url
+
+
 def get_database_url():
     """Obtiene la URL de la base de datos (PostgreSQL) desde variables de entorno o alembic.ini."""
-    # Intentar construir desde variables de entorno primero
+    # CI / pytest: misma URL que los tests (secret TEST_DATABASE_URL) o DATABASE_URL explícita
+    for env_key in ("TEST_DATABASE_URL"):
+        raw = os.environ.get(env_key)
+        if raw and raw.strip():
+            return _normalize_alembic_sqlalchemy_url(raw)
+
+    # Construcción desde settings (CORE_DB_* en .env / entorno)
     host = settings.CORE_DB_HOST
     db = settings.CORE_DB_NAME
     user = settings.CORE_DB_USER
@@ -41,7 +61,8 @@ def get_database_url():
     if not url:
         raise RuntimeError(
             "No se ha configurado la cadena de conexión a PostgreSQL. "
-            "Proporciona las variables de entorno POSTGRES_* o configura sqlalchemy.url en alembic.ini"
+            "Define TEST_DATABASE_URL o DATABASE_URL, o CORE_DB_HOST/CORE_DB_USER/CORE_DB_PASSWORD/CORE_DB_NAME, "
+            "o sqlalchemy.url en alembic.ini"
         )
 
     return url

From a3ba317d2cc4ceb4ccbe05a3e06f74ade0cf71e9 Mon Sep 17 00:00:00 2001
From: AlexeerCT <acazares@aduanasoft.com.mx>
Date: Tue, 24 Mar 2026 15:32:49 -0500
Subject: [PATCH 03/10] Enhance database URL handling in Alembic environment

- Added a new helper function to strip unwanted characters from environment variable URLs, improving the normalization process.
- Updated the `get_database_url` function to validate the database URL scheme and provide clearer error messages for misconfigurations.
- Expanded the environment variable checks to include `DATABASE_URL`, enhancing flexibility for different deployment scenarios.

These changes aim to improve the robustness and clarity of database connection handling in the application.
---
 backend/alembic/env.py | 27 ++++++++++++++++++++-------
 1 file changed, 20 insertions(+), 7 deletions(-)

diff --git a/backend/alembic/env.py b/backend/alembic/env.py
index bfbd92f1..2905babf 100644
--- a/backend/alembic/env.py
+++ b/backend/alembic/env.py
@@ -3,7 +3,7 @@ import logging
 import os
 import sys
 from logging.config import fileConfig
-from urllib.parse import quote_plus
+from urllib.parse import quote_plus, urlparse
 
 from alembic import context
 from alembic.operations import ops
@@ -18,9 +18,15 @@ logger = logging.getLogger(__name__)
 config = context.config
 
 
+def _strip_env_url(raw: str) -> str:
+    """Quita espacios/comillas típicos de secretos CI (.env, Gitea)."""
+    url = raw.strip().strip('"').strip("'")
+    return url
+
+
 def _normalize_alembic_sqlalchemy_url(url: str) -> str:
     """Alembic usa el driver síncrono psycopg2; normaliza DSN típicos de app/tests."""
-    url = url.strip()
+    url = _strip_env_url(url)
     if url.startswith("postgresql+asyncpg://"):
         return url.replace("postgresql+asyncpg://", "postgresql+psycopg2://", 1)
     if url.startswith("postgresql+psycopg2://"):
@@ -34,11 +40,18 @@ def _normalize_alembic_sqlalchemy_url(url: str) -> str:
 
 def get_database_url():
     """Obtiene la URL de la base de datos (PostgreSQL) desde variables de entorno o alembic.ini."""
-    # CI / pytest: misma URL que los tests (secret TEST_DATABASE_URL) o DATABASE_URL explícita
+    # CI / pytest: misma URL que los tests (secret TEST_DATABASE_URL) o DATABASE_URL explícita.
+    # Nota: ("X") sin coma es str, no tupla; el for iteraría caracteres y jamás leería la variable.
     for env_key in ("TEST_DATABASE_URL"):
         raw = os.environ.get(env_key)
         if raw and raw.strip():
-            return _normalize_alembic_sqlalchemy_url(raw)
+            normalized = _normalize_alembic_sqlalchemy_url(raw)
+            if not urlparse(normalized).scheme:
+                raise RuntimeError(
+                    f"{env_key} no es una URL válida (falta esquema). "
+                    "Ejemplo: postgresql://usuario:clave@host:5432/nombre_bd"
+                )
+            return normalized
 
     # Construcción desde settings (CORE_DB_* en .env / entorno)
     host = settings.CORE_DB_HOST
@@ -58,11 +71,11 @@ def get_database_url():
 
     # Fallback al archivo de configuración
     url = config.get_main_option("sqlalchemy.url")
-    if not url:
+    if not url or "${" in url or "%(" in url:
         raise RuntimeError(
             "No se ha configurado la cadena de conexión a PostgreSQL. "
-            "Define TEST_DATABASE_URL o DATABASE_URL, o CORE_DB_HOST/CORE_DB_USER/CORE_DB_PASSWORD/CORE_DB_NAME, "
-            "o sqlalchemy.url en alembic.ini"
+            "Define TEST_DATABASE_URL o DATABASE_URL, o variables CORE_DB_*; "
+            "sqlalchemy.url en alembic.ini con placeholders ${...} no está soportado."
         )
 
     return url

From 18e4477bf120b3ff3d6a4dd953ae1a7d2a25acb0 Mon Sep 17 00:00:00 2001
From: AlexeerCT <acazares@aduanasoft.com.mx>
Date: Tue, 24 Mar 2026 15:35:13 -0500
Subject: [PATCH 04/10] Add TEST_DATABASE_URL to configuration settings

- Introduced a new configuration variable, TEST_DATABASE_URL, to specify the test database connection string.
- This addition enhances the flexibility of database handling, particularly for testing scenarios, aligning with recent improvements in database URL management.
---
 backend/core/config.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/backend/core/config.py b/backend/core/config.py
index 495600f8..eaabdb82 100644
--- a/backend/core/config.py
+++ b/backend/core/config.py
@@ -27,6 +27,8 @@ class Settings(BaseSettings):
     CORE_DB_USER: str = "postgres"
     CORE_DB_PASSWORD: str = "postgres"
 
+    TEST_DATABASE_URL: str = "postgresql://postgres:postgres@localhost:5432/anexo76_core"
+
     # Keycloak
     KEYCLOAK_SERVER_URL: str = "http://localhost:8080/kcauth"
     KEYCLOAK_REALM: str = "master"

From 2cd6b165d789fbf06f00fb7dd23f15d37eb71350 Mon Sep 17 00:00:00 2001
From: AlexeerCT <acazares@aduanasoft.com.mx>
Date: Tue, 24 Mar 2026 15:39:53 -0500
Subject: [PATCH 05/10] Enhance database URL validation in Alembic environment

- Introduced a new validation function for SQLAlchemy URLs to ensure proper formatting and provide clearer error messages.
- Updated the `get_database_url` function to include checks for both `TEST_DATABASE_URL` and `DATABASE_URL`, improving flexibility in configuration.
- Removed the previous URL scheme validation in favor of the new validation method, streamlining the error handling process.

These changes aim to improve the robustness and clarity of database connection handling in the application.
---
 backend/alembic/env.py | 25 +++++++++++++++++--------
 1 file changed, 17 insertions(+), 8 deletions(-)

diff --git a/backend/alembic/env.py b/backend/alembic/env.py
index 2905babf..6b0d8d8c 100644
--- a/backend/alembic/env.py
+++ b/backend/alembic/env.py
@@ -3,13 +3,14 @@ import logging
 import os
 import sys
 from logging.config import fileConfig
-from urllib.parse import quote_plus, urlparse
+from urllib.parse import quote_plus
 
 from alembic import context
 from alembic.operations import ops
 from core.config import settings
 from core.database import Base
 from sqlalchemy import engine_from_config, pool
+from sqlalchemy.engine.url import make_url
 
 logger = logging.getLogger(__name__)
 
@@ -38,19 +39,27 @@ def _normalize_alembic_sqlalchemy_url(url: str) -> str:
     return url
 
 
+def _validate_sqlalchemy_url(url: str, env_key: str) -> None:
+    """Misma validación que create_engine; evita urlparse (falla con esquemas tipo postgresql+psycopg2)."""
+    try:
+        make_url(url)
+    except Exception as e:
+        raise RuntimeError(
+            f"{env_key} no es una URL de SQLAlchemy válida. "
+            "Ejemplo: postgresql://usuario:clave@host:5432/nombre_bd"
+        ) from e
+
+
 def get_database_url():
     """Obtiene la URL de la base de datos (PostgreSQL) desde variables de entorno o alembic.ini."""
     # CI / pytest: misma URL que los tests (secret TEST_DATABASE_URL) o DATABASE_URL explícita.
-    # Nota: ("X") sin coma es str, no tupla; el for iteraría caracteres y jamás leería la variable.
-    for env_key in ("TEST_DATABASE_URL"):
+    # CRÍTICO: debe ser tupla con coma final si un solo elemento: ("X",) — si no, ("X") es str y el for
+    # itera caracteres; env_key "_" + os.environ["_"] (común en shells) rompe con URL inválida.
+    for env_key in ("TEST_DATABASE_URL", "DATABASE_URL"):
         raw = os.environ.get(env_key)
         if raw and raw.strip():
             normalized = _normalize_alembic_sqlalchemy_url(raw)
-            if not urlparse(normalized).scheme:
-                raise RuntimeError(
-                    f"{env_key} no es una URL válida (falta esquema). "
-                    "Ejemplo: postgresql://usuario:clave@host:5432/nombre_bd"
-                )
+            _validate_sqlalchemy_url(normalized, env_key)
             return normalized
 
     # Construcción desde settings (CORE_DB_* en .env / entorno)

From 309835be8cd219f1541381198ce395c92972e8e6 Mon Sep 17 00:00:00 2001
From: AlexeerCT <acazares@aduanasoft.com.mx>
Date: Tue, 24 Mar 2026 15:41:47 -0500
Subject: [PATCH 06/10] Enhance SQLAlchemy URL validation in Alembic
 environment

- Added a new function to reject documentation placeholder hosts in database URLs, improving error handling for misconfigurations.
- Updated the `get_database_url` function to incorporate this new validation, ensuring that users are alerted when using "host" as a placeholder.
- Enhanced error messages to provide clearer guidance on valid host configurations.

These changes aim to improve the robustness and clarity of database connection handling in the application.
---
 backend/alembic/env.py  | 25 +++++++++++++++++++++++--
 backend/tests/README.md |  2 +-
 2 files changed, 24 insertions(+), 3 deletions(-)

diff --git a/backend/alembic/env.py b/backend/alembic/env.py
index 6b0d8d8c..07823c7c 100644
--- a/backend/alembic/env.py
+++ b/backend/alembic/env.py
@@ -46,10 +46,27 @@ def _validate_sqlalchemy_url(url: str, env_key: str) -> None:
     except Exception as e:
         raise RuntimeError(
             f"{env_key} no es una URL de SQLAlchemy válida. "
-            "Ejemplo: postgresql://usuario:clave@host:5432/nombre_bd"
+            "Ejemplo: postgresql://usuario:clave@127.0.0.1:5432/nombre_bd"
         ) from e
 
 
+def _reject_documentation_placeholder_host(url: str, source: str) -> None:
+    """
+    Evita el error críptico de DNS: muchos ejemplos usan @host:5432 como texto literal.
+    """
+    try:
+        parsed = make_url(url)
+    except Exception:
+        return
+    h = (parsed.host or "").strip().lower()
+    if h == "host":
+        raise RuntimeError(
+            f"{source}: el hostname \"host\" es un placeholder de documentación, no un servidor real. "
+            "Usa el host alcanzable desde el runner (IP, nombre DNS, servicio en docker-compose, "
+            "o host.docker.internal si act corre en contenedor y Postgres en tu máquina)."
+        )
+
+
 def get_database_url():
     """Obtiene la URL de la base de datos (PostgreSQL) desde variables de entorno o alembic.ini."""
     # CI / pytest: misma URL que los tests (secret TEST_DATABASE_URL) o DATABASE_URL explícita.
@@ -60,6 +77,7 @@ def get_database_url():
         if raw and raw.strip():
             normalized = _normalize_alembic_sqlalchemy_url(raw)
             _validate_sqlalchemy_url(normalized, env_key)
+            _reject_documentation_placeholder_host(normalized, env_key)
             return normalized
 
     # Construcción desde settings (CORE_DB_* en .env / entorno)
@@ -74,7 +92,9 @@ def get_database_url():
             encoded_user = quote_plus(user)
             encoded_password = quote_plus(password)
             encoded_db = quote_plus(db)
-            return f"postgresql+psycopg2://{encoded_user}:{encoded_password}@{host}:{port}/{encoded_db}"
+            built = f"postgresql+psycopg2://{encoded_user}:{encoded_password}@{host}:{port}/{encoded_db}"
+            _reject_documentation_placeholder_host(built, "CORE_DB_HOST")
+            return built
         except Exception as e:
             logger.error(f"Error al construir URL: {e}")
 
@@ -87,6 +107,7 @@ def get_database_url():
             "sqlalchemy.url en alembic.ini con placeholders ${...} no está soportado."
         )
 
+    _reject_documentation_placeholder_host(url, "alembic.ini sqlalchemy.url")
     return url
 
 
diff --git a/backend/tests/README.md b/backend/tests/README.md
index ef790e4b..6e327370 100644
--- a/backend/tests/README.md
+++ b/backend/tests/README.md
@@ -134,7 +134,7 @@ Si `TEST_DATABASE_URL` apunta a una base con datos previos, los builders reutili
 
 - PostgreSQL test database available.
 - Environment variable:
-  - `TEST_DATABASE_URL=postgresql://user:pass@host:5432/db_name`
+  - `TEST_DATABASE_URL=postgresql://user:pass@127.0.0.1:5432/db_name` (sustituye por el host real alcanzable desde el runner; no uses el literal `host` de los ejemplos genéricos)
 
 ## Run commands
 

From 1150b59dcf08641722da3585ccb7e33caa29e6a8 Mon Sep 17 00:00:00 2001
From: AlexeerCT <acazares@aduanasoft.com.mx>
Date: Tue, 24 Mar 2026 15:42:22 -0500
Subject: [PATCH 07/10] Update example for TEST_DATABASE_URL in CI workflow

- Modified the error message in the build workflow to clarify the expected format for the TEST_DATABASE_URL, specifying the use of a real IP address instead of the placeholder "host".
- This change enhances the guidance provided to users, ensuring correct configuration for database connections in testing scenarios.
---
 .gitea/workflows/build.yml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/.gitea/workflows/build.yml b/.gitea/workflows/build.yml
index 23a2f909..24d77a37 100644
--- a/.gitea/workflows/build.yml
+++ b/.gitea/workflows/build.yml
@@ -36,7 +36,7 @@ jobs:
           set -e
           if [ -z "$TEST_DATABASE_URL" ]; then
             echo "::error::Define el secret TEST_DATABASE_URL en el repo (Gitea → Ajustes → Secretos)."
-            echo "Ejemplo: postgresql://usuario:clave@host:5432/nombre_bd"
+            echo "Ejemplo: postgresql://usuario:clave@127.0.0.1:5432/nombre_bd (host real, no el texto \"host\")"
             exit 1
           fi
           python3 --version

From 06b3b20fd824694a918c1d7d62e920c1cd18a222 Mon Sep 17 00:00:00 2001
From: AlexeerCT <acazares@aduanasoft.com.mx>
Date: Tue, 24 Mar 2026 17:24:01 -0500
Subject: [PATCH 08/10] Refactor test setup to utilize ephemeral tenant IDs

- Introduced a new fixture, `test_tenant`, to allocate ephemeral tenant and company IDs for tests, reducing conflicts with real data.
- Updated various test functions to use the new `test_tenant` fixture, ensuring consistent tenant ID usage across tests.
- Enhanced the `ensure_tenant_company` function to accept dynamic tenant and company IDs, improving flexibility in test scenarios.
- Adjusted database interaction in tests to utilize the ephemeral IDs, streamlining the setup process and enhancing isolation.
---
 .../versions/bccb7f8986c7_iva_factor.py       | 42 +++++++++++++++--
 backend/tests/README.md                       |  4 +-
 backend/tests/conftest.py                     | 21 ++++++++-
 .../tests/e2e/test_inventory_flow_anexo24.py  | 16 +++----
 backend/tests/fixtures/builders.py            | 31 ++++++++++++-
 .../integration/test_export_process_api.py    | 46 +++++++++++--------
 .../integration/test_import_process_api.py    | 21 ++++-----
 .../unit/invoices/test_balance_algorithm.py   | 11 ++---
 8 files changed, 136 insertions(+), 56 deletions(-)

diff --git a/backend/alembic/versions/bccb7f8986c7_iva_factor.py b/backend/alembic/versions/bccb7f8986c7_iva_factor.py
index 2c31ffb5..5a9d0d34 100644
--- a/backend/alembic/versions/bccb7f8986c7_iva_factor.py
+++ b/backend/alembic/versions/bccb7f8986c7_iva_factor.py
@@ -18,11 +18,31 @@ branch_labels: Union[str, Sequence[str], None] = None
 depends_on: Union[str, Sequence[str], None] = None
 
 
+def _public_table_exists(bind, name: str) -> bool:
+    """Evita CREATE INDEX si la tabla no existe (BD parcial / orden atípico de migraciones)."""
+    return name in sa.inspect(bind).get_table_names(schema="public")
+
+
 def upgrade() -> None:
     """Upgrade schema."""
     # ### commands auto generated by Alembic - please adjust! ###
-    op.drop_index(op.f('ix_public_carta_porte_code'), table_name='carta_porte_codes')
-    op.create_index(op.f('ix_public_carta_porte_codes_code'), 'carta_porte_codes', ['code'], unique=False, schema='public')
+    bind = op.get_bind()
+    if _public_table_exists(bind, "carta_porte_codes"):
+        # Coincide con 4ad64605fad2 (schema=public). if_exists: sin índice previo.
+        op.drop_index(
+            op.f("ix_public_carta_porte_code"),
+            table_name="carta_porte_codes",
+            schema="public",
+            if_exists=True,
+        )
+        op.create_index(
+            op.f("ix_public_carta_porte_codes_code"),
+            "carta_porte_codes",
+            ["code"],
+            unique=False,
+            schema="public",
+            if_not_exists=True,
+        )
     op.alter_column('invoice_financials', 'iva_factor',
                existing_type=sa.VARCHAR(length=10),
                type_=sa.Numeric(precision=23, scale=8),
@@ -40,6 +60,20 @@ def downgrade() -> None:
                type_=sa.VARCHAR(length=10),
                existing_nullable=True,
                schema='a76')
-    op.drop_index(op.f('ix_public_carta_porte_codes_code'), table_name='carta_porte_codes', schema='public')
-    op.create_index(op.f('ix_public_carta_porte_code'), 'carta_porte_codes', ['code'], unique=False)
+    bind = op.get_bind()
+    if _public_table_exists(bind, "carta_porte_codes"):
+        op.drop_index(
+            op.f("ix_public_carta_porte_codes_code"),
+            table_name="carta_porte_codes",
+            schema="public",
+            if_exists=True,
+        )
+        op.create_index(
+            op.f("ix_public_carta_porte_code"),
+            "carta_porte_codes",
+            ["code"],
+            unique=False,
+            schema="public",
+            if_not_exists=True,
+        )
     # ### end Alembic commands ###
diff --git a/backend/tests/README.md b/backend/tests/README.md
index 6e327370..ad75670f 100644
--- a/backend/tests/README.md
+++ b/backend/tests/README.md
@@ -44,14 +44,14 @@ flowchart TB
 
 - **BD**: `TEST_DATABASE_URL` o `CORE_DATABASE_URL` o `settings.core_database_url`. Cada test corre dentro de una transacción; al terminar **no persiste** nada (rollback).
 - **Celery**: sin Redis; `task_always_eager` para que el endpoint dispare lógica síncrona.
-- **Seguridad en tests**: `validate_access_to_resource` se reemplaza por un stub que devuelve `tenant_id` del usuario fake.
+- **Seguridad en tests**: `validate_access_to_resource` se reemplaza por un stub que devuelve `tenant_id` del usuario fake (alineado con el fixture `test_tenant`).
 
 ### Builders: qué se construye antes de llamar a la API
 
 Orden típico en casi todos los tests que tocan proceso:
 
 1. `ensure_reference_data`: tipos de factura (`TEM`, `DEF`, `MEX`, `DONAC`) y régimen `A1` si no existen.
-2. `ensure_tenant_company(tenant_id=1, company_id=1)`.
+2. Fixture `test_tenant`: asigna IDs efímeros altos y crea tenant + company; el `app` usa ese `tenant_id` en el usuario fake.
 3. `create_business_catalogs`: proveedor/cliente/destinatario (con dirección idempotente vía `_ensure_client_provider_address`), agente aduanal, UoM, part number FA, tipo cambio, etc.
 4. `create_import_invoice_with_line` / `create_export_invoice_with_line` según el caso.
 
diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py
index 1be795d3..3f601f19 100644
--- a/backend/tests/conftest.py
+++ b/backend/tests/conftest.py
@@ -17,6 +17,11 @@ from core.celery_app import celery_app
 from core.config import settings
 from core.database import get_core_db
 from core.security import get_current_user
+from tests.fixtures.builders import (
+    EphemeralOrgIds,
+    allocate_ephemeral_tenant_company_ids,
+    ensure_tenant_company,
+)
 
 
 TEST_DB_URL = (
@@ -49,6 +54,14 @@ def db_session() -> Generator[Session, None, None]:
         connection.close()
 
 
+@pytest.fixture
+def test_tenant(db_session: Session) -> EphemeralOrgIds:
+    """Tenant y company dedicados por test (IDs efímeros en rango alto)."""
+    tid, cid = allocate_ephemeral_tenant_company_ids(db_session)
+    ensure_tenant_company(db_session, tenant_id=tid, company_id=cid)
+    return EphemeralOrgIds(tenant_id=tid, company_id=cid)
+
+
 @pytest.fixture(autouse=True)
 def celery_eager() -> Generator[None, None, None]:
     prev_broker = celery_app.conf.broker_url
@@ -77,7 +90,11 @@ def celery_eager() -> Generator[None, None, None]:
 
 
 @pytest.fixture
-def app(db_session: Session, monkeypatch: pytest.MonkeyPatch) -> FastAPI:
+def app(
+    db_session: Session,
+    test_tenant: EphemeralOrgIds,
+    monkeypatch: pytest.MonkeyPatch,
+) -> FastAPI:
     test_app = FastAPI()
     test_app.include_router(process_routes.router, prefix="/api/v1/a76")
 
@@ -85,7 +102,7 @@ def app(db_session: Session, monkeypatch: pytest.MonkeyPatch) -> FastAPI:
         yield db_session
 
     async def _override_current_user():
-        return {"sub": "test-user", "tenant_id": 1}
+        return {"sub": "test-user", "tenant_id": test_tenant.tenant_id}
 
     # validate_access_to_resource is imported directly in the routes module.
     monkeypatch.setattr(
diff --git a/backend/tests/e2e/test_inventory_flow_anexo24.py b/backend/tests/e2e/test_inventory_flow_anexo24.py
index 19be14b6..0daab7a8 100644
--- a/backend/tests/e2e/test_inventory_flow_anexo24.py
+++ b/backend/tests/e2e/test_inventory_flow_anexo24.py
@@ -4,13 +4,11 @@ from api.v1.modules.a24.balance_movements.models import BalanceMovement, Movemen
 from api.v1.modules.a24.discharges.models import DischargeDetail, DischargeHeader
 from api.v1.modules.a76.invoices.imports.process import main_process as import_main
 from api.v1.modules.a76.invoices.exports.process import main_process as export_main
-from api.v1.modules.a76.invoices.exports.process import task as export_task
 from tests.fixtures.builders import (
     create_business_catalogs,
     create_export_invoice_with_line,
     create_import_invoice_with_line,
     ensure_reference_data,
-    ensure_tenant_company,
 )
 
 
@@ -34,7 +32,7 @@ def _patch_export_pipeline(monkeypatch, export_line):
     monkeypatch.setattr(export_main, "review_qty_series", lambda *args, **kwargs: None)
 
 
-def test_e2e_inventory_flow_import_then_export(client, db_session, monkeypatch):
+def test_e2e_inventory_flow_import_then_export(client, db_session, monkeypatch, test_tenant):
     """
     E2E principal:
     - Alta catálogos/base
@@ -42,16 +40,16 @@ def test_e2e_inventory_flow_import_then_export(client, db_session, monkeypatch):
     - Procesa factura EXP -> genera consumo/descarga
     - Valida decremento de saldo sin negativos
     """
+    tid, cid = test_tenant.tenant_id, test_tenant.company_id
     ensure_reference_data(db_session)
-    ensure_tenant_company(db_session, tenant_id=1, company_id=1)
-    catalogs = create_business_catalogs(db_session, tenant_id=1, company_id=1)
+    catalogs = create_business_catalogs(db_session, tenant_id=tid, company_id=cid)
 
     import_invoice, import_line = create_import_invoice_with_line(
-        db_session, 1, 1, catalogs, invoice_type="TEM", invoice_number="IMP-E2E-01", qty=Decimal("10")
+        db_session, tid, cid, catalogs, invoice_type="TEM", invoice_number="IMP-E2E-01", qty=Decimal("10")
     )
     _patch_import_pipeline(monkeypatch, import_line)
 
-    response = client.post(f"/api/v1/a76/invoices/{import_invoice.id}/process?company_id=1")
+    response = client.post(f"/api/v1/a76/invoices/{import_invoice.id}/process?company_id={cid}")
     assert response.status_code == 200
 
     entry_movements = (
@@ -67,11 +65,11 @@ def test_e2e_inventory_flow_import_then_export(client, db_session, monkeypatch):
     assert sum(Decimal(str(m.quantity or 0)) for m in entry_movements) > 0
 
     export_invoice, export_line = create_export_invoice_with_line(
-        db_session, 1, 1, catalogs, import_invoice, import_line, qty=Decimal("4")
+        db_session, tid, cid, catalogs, import_invoice, import_line, qty=Decimal("4")
     )
     _patch_export_pipeline(monkeypatch, export_line)
 
-    response = client.post(f"/api/v1/a76/invoices/{export_invoice.id}/process?company_id=1")
+    response = client.post(f"/api/v1/a76/invoices/{export_invoice.id}/process?company_id={cid}")
     assert response.status_code == 200
 
     consumptions = (
diff --git a/backend/tests/fixtures/builders.py b/backend/tests/fixtures/builders.py
index d1f0671d..1e836dd0 100644
--- a/backend/tests/fixtures/builders.py
+++ b/backend/tests/fixtures/builders.py
@@ -1,7 +1,9 @@
 from __future__ import annotations
 
+import uuid
 from datetime import date, datetime
 from decimal import Decimal
+from typing import NamedTuple
 
 from sqlalchemy.orm import Session
 
@@ -34,6 +36,31 @@ from api.v1.modules.public.reference_data.invoice_types.models import InvoiceTyp
 from api.v1.modules.public.reference_data.pedimento_regimens.models import RegimenPedimento
 
 
+class EphemeralOrgIds(NamedTuple):
+    """Tenant + company creados por test (IDs altos, baja colisión con datos reales / CI)."""
+
+    tenant_id: int
+    company_id: int
+
+
+def allocate_ephemeral_tenant_company_ids(db: Session) -> tuple[int, int]:
+    """
+    Genera par (tenant_id, company_id) no usados en esta sesión.
+    Mismo entero para ambas PKs es válido (tablas distintas).
+    """
+    for _ in range(64):
+        n = uuid.uuid4().int
+        tid = 1_500_000_000 + (n % 99_000_000)
+        if db.get(Tenant, tid) is not None:
+            continue
+        if db.get(Company, tid) is not None:
+            continue
+        return tid, tid
+    raise RuntimeError(
+        "No se pudo asignar tenant/company efímero para tests (reintentar o revisar datos en BD)."
+    )
+
+
 def ensure_reference_data(db: Session) -> None:
     for key, desc in [
         ("TEM", "IMPORTACION TEMPORAL"),
@@ -63,13 +90,13 @@ def ensure_reference_data(db: Session) -> None:
     db.flush()
 
 
-def ensure_tenant_company(db: Session, tenant_id: int = 1, company_id: int = 1) -> Company:
+def ensure_tenant_company(db: Session, tenant_id: int, company_id: int) -> Company:
     tenant = db.get(Tenant, tenant_id)
     if tenant is None:
         tenant = Tenant(
             id=tenant_id,
             name=f"Tenant {tenant_id}",
-            slug=f"tenant-{tenant_id}",
+            slug=f"pytest-{tenant_id}-{uuid.uuid4().hex[:10]}",
             type=TenantType.SHARED,
             keycloak_realm="test",
             is_active=True,
diff --git a/backend/tests/integration/test_export_process_api.py b/backend/tests/integration/test_export_process_api.py
index 411a894c..ed51d68b 100644
--- a/backend/tests/integration/test_export_process_api.py
+++ b/backend/tests/integration/test_export_process_api.py
@@ -7,11 +7,11 @@ from api.v1.modules.a76.invoices.exports.process import main_process as export_m
 from api.v1.modules.a76.invoices.imports.process import main_process as import_main
 from core.exceptions import ValidationException
 from tests.fixtures.builders import (
+    EphemeralOrgIds,
     create_business_catalogs,
     create_export_invoice_with_line,
     create_import_invoice_with_line,
     ensure_reference_data,
-    ensure_tenant_company,
 )
 
 
@@ -44,27 +44,30 @@ def _patch_export_pipeline(monkeypatch, export_line):
     monkeypatch.setattr(export_main, "review_qty_series", lambda *args, **kwargs: None)
 
 
-def _prepare_inventory(client, db_session, monkeypatch):
+def _prepare_inventory(client, db_session, monkeypatch, test_tenant: EphemeralOrgIds):
+    tid, cid = test_tenant.tenant_id, test_tenant.company_id
     ensure_reference_data(db_session)
-    ensure_tenant_company(db_session, tenant_id=1, company_id=1)
-    catalogs = create_business_catalogs(db_session, tenant_id=1, company_id=1)
+    catalogs = create_business_catalogs(db_session, tenant_id=tid, company_id=cid)
     import_invoice, import_line = create_import_invoice_with_line(
-        db_session, 1, 1, catalogs, invoice_type="TEM", invoice_number="IMP-EXP-01", qty=Decimal("10")
+        db_session, tid, cid, catalogs, invoice_type="TEM", invoice_number="IMP-EXP-01", qty=Decimal("10")
     )
     _patch_import_pipeline(monkeypatch, import_line)
-    resp = client.post(f"/api/v1/a76/invoices/{import_invoice.id}/process?company_id=1")
+    resp = client.post(f"/api/v1/a76/invoices/{import_invoice.id}/process?company_id={cid}")
     assert resp.status_code == 200
     return catalogs, import_invoice, import_line
 
 
-def test_process_export_endpoint_consumes_existing_balances(client, db_session, monkeypatch):
-    catalogs, import_invoice, import_line = _prepare_inventory(client, db_session, monkeypatch)
+def test_process_export_endpoint_consumes_existing_balances(client, db_session, monkeypatch, test_tenant):
+    catalogs, import_invoice, import_line = _prepare_inventory(
+        client, db_session, monkeypatch, test_tenant
+    )
+    tid, cid = test_tenant.tenant_id, test_tenant.company_id
     export_invoice, export_line = create_export_invoice_with_line(
-        db_session, 1, 1, catalogs, import_invoice, import_line, qty=Decimal("3")
+        db_session, tid, cid, catalogs, import_invoice, import_line, qty=Decimal("3")
     )
     _patch_export_pipeline(monkeypatch, export_line)
 
-    resp = client.post(f"/api/v1/a76/invoices/{export_invoice.id}/process?company_id=1")
+    resp = client.post(f"/api/v1/a76/invoices/{export_invoice.id}/process?company_id={cid}")
     assert resp.status_code == 200
 
     consumptions = (
@@ -78,15 +81,18 @@ def test_process_export_endpoint_consumes_existing_balances(client, db_session,
     assert consumptions
 
 
-def test_process_export_prevents_negative_balance(client, db_session, monkeypatch):
-    catalogs, import_invoice, import_line = _prepare_inventory(client, db_session, monkeypatch)
+def test_process_export_prevents_negative_balance(client, db_session, monkeypatch, test_tenant):
+    catalogs, import_invoice, import_line = _prepare_inventory(
+        client, db_session, monkeypatch, test_tenant
+    )
+    tid, cid = test_tenant.tenant_id, test_tenant.company_id
     export_invoice, export_line = create_export_invoice_with_line(
-        db_session, 1, 1, catalogs, import_invoice, import_line, qty=Decimal("99")
+        db_session, tid, cid, catalogs, import_invoice, import_line, qty=Decimal("99")
     )
     _patch_export_pipeline(monkeypatch, export_line)
 
     with pytest.raises(ValidationException):
-        client.post(f"/api/v1/a76/invoices/{export_invoice.id}/process?company_id=1")
+        client.post(f"/api/v1/a76/invoices/{export_invoice.id}/process?company_id={cid}")
 
     # Debe no crear consumos al existir insuficiencia de saldo.
     consumptions = (
@@ -100,19 +106,19 @@ def test_process_export_prevents_negative_balance(client, db_session, monkeypatc
     assert consumptions == []
 
 
-def test_process_endpoint_prevents_double_processing_import(client, db_session, monkeypatch):
+def test_process_endpoint_prevents_double_processing_import(client, db_session, monkeypatch, test_tenant):
+    tid, cid = test_tenant.tenant_id, test_tenant.company_id
     ensure_reference_data(db_session)
-    ensure_tenant_company(db_session, tenant_id=1, company_id=1)
-    catalogs = create_business_catalogs(db_session, tenant_id=1, company_id=1)
+    catalogs = create_business_catalogs(db_session, tenant_id=tid, company_id=cid)
     invoice, line = create_import_invoice_with_line(
-        db_session, 1, 1, catalogs, invoice_type="TEM", invoice_number="IMP-DOUBLE-01", qty=Decimal("5")
+        db_session, tid, cid, catalogs, invoice_type="TEM", invoice_number="IMP-DOUBLE-01", qty=Decimal("5")
     )
     _patch_import_pipeline_without_prevalidators(monkeypatch)
 
-    first = client.post(f"/api/v1/a76/invoices/{invoice.id}/process?company_id=1")
+    first = client.post(f"/api/v1/a76/invoices/{invoice.id}/process?company_id={cid}")
     assert first.status_code == 200
     with pytest.raises(ValidationException):
-        client.post(f"/api/v1/a76/invoices/{invoice.id}/process?company_id=1")
+        client.post(f"/api/v1/a76/invoices/{invoice.id}/process?company_id={cid}")
 
     # No debe duplicar entradas para la misma factura/línea procesada.
     entries = (
diff --git a/backend/tests/integration/test_import_process_api.py b/backend/tests/integration/test_import_process_api.py
index fd0aadaa..956f759c 100644
--- a/backend/tests/integration/test_import_process_api.py
+++ b/backend/tests/integration/test_import_process_api.py
@@ -6,7 +6,6 @@ from tests.fixtures.builders import (
     create_business_catalogs,
     create_import_invoice_with_line,
     ensure_reference_data,
-    ensure_tenant_company,
 )
 
 
@@ -20,16 +19,16 @@ def _patch_import_pipeline(monkeypatch, import_line):
     monkeypatch.setattr(import_main, "_validate_lines", lambda *args, **kwargs: ([], {}))
 
 
-def test_process_import_endpoint_creates_balance_entries(client, db_session, monkeypatch):
+def test_process_import_endpoint_creates_balance_entries(client, db_session, monkeypatch, test_tenant):
+    tid, cid = test_tenant.tenant_id, test_tenant.company_id
     ensure_reference_data(db_session)
-    ensure_tenant_company(db_session, tenant_id=1, company_id=1)
-    catalogs = create_business_catalogs(db_session, tenant_id=1, company_id=1)
+    catalogs = create_business_catalogs(db_session, tenant_id=tid, company_id=cid)
     invoice, line = create_import_invoice_with_line(
-        db_session, 1, 1, catalogs, invoice_type="TEM", invoice_number="IMP-INT-01", qty=Decimal("7")
+        db_session, tid, cid, catalogs, invoice_type="TEM", invoice_number="IMP-INT-01", qty=Decimal("7")
     )
     _patch_import_pipeline(monkeypatch, line)
 
-    resp = client.post(f"/api/v1/a76/invoices/{invoice.id}/process?company_id=1")
+    resp = client.post(f"/api/v1/a76/invoices/{invoice.id}/process?company_id={cid}")
     assert resp.status_code == 200
 
     entries = (
@@ -44,16 +43,16 @@ def test_process_import_endpoint_creates_balance_entries(client, db_session, mon
     assert len(entries) >= 1
 
 
-def test_process_import_def_does_not_create_balance_entries(client, db_session, monkeypatch):
+def test_process_import_def_does_not_create_balance_entries(client, db_session, monkeypatch, test_tenant):
+    tid, cid = test_tenant.tenant_id, test_tenant.company_id
     ensure_reference_data(db_session)
-    ensure_tenant_company(db_session, tenant_id=1, company_id=1)
-    catalogs = create_business_catalogs(db_session, tenant_id=1, company_id=1)
+    catalogs = create_business_catalogs(db_session, tenant_id=tid, company_id=cid)
     invoice, line = create_import_invoice_with_line(
-        db_session, 1, 1, catalogs, invoice_type="DEF", invoice_number="IMP-DEF-01", qty=Decimal("7")
+        db_session, tid, cid, catalogs, invoice_type="DEF", invoice_number="IMP-DEF-01", qty=Decimal("7")
     )
     _patch_import_pipeline(monkeypatch, line)
 
-    resp = client.post(f"/api/v1/a76/invoices/{invoice.id}/process?company_id=1")
+    resp = client.post(f"/api/v1/a76/invoices/{invoice.id}/process?company_id={cid}")
     assert resp.status_code == 200
 
     entries = (
diff --git a/backend/tests/unit/invoices/test_balance_algorithm.py b/backend/tests/unit/invoices/test_balance_algorithm.py
index a158ed06..70b6c1df 100644
--- a/backend/tests/unit/invoices/test_balance_algorithm.py
+++ b/backend/tests/unit/invoices/test_balance_algorithm.py
@@ -15,22 +15,21 @@ from tests.fixtures.builders import (
     create_business_catalogs,
     create_import_invoice_with_line,
     ensure_reference_data,
-    ensure_tenant_company,
 )
 
 
-def test_net_balance_accounts_for_returns_and_entry_void(db_session):
+def test_net_balance_accounts_for_returns_and_entry_void(db_session, test_tenant):
     """
     Balance neto esperado:
       entry 10 - consumption 4 + return 1 - entry_void 2 = 5
     """
+    tid, cid = test_tenant.tenant_id, test_tenant.company_id
     ensure_reference_data(db_session)
-    ensure_tenant_company(db_session, tenant_id=1, company_id=1)
-    catalogs = create_business_catalogs(db_session, tenant_id=1, company_id=1)
+    catalogs = create_business_catalogs(db_session, tenant_id=tid, company_id=cid)
     import_invoice, import_line = create_import_invoice_with_line(
         db_session,
-        1,
-        1,
+        tid,
+        cid,
         catalogs,
         invoice_type="TEM",
         invoice_number="IMP-UNIT-01",

From 28de77e50b69a97634e3784429af741f2218ed46 Mon Sep 17 00:00:00 2001
From: AlexeerCT <acazares@aduanasoft.com.mx>
Date: Tue, 24 Mar 2026 18:05:24 -0500
Subject: [PATCH 09/10] Refactor Alembic migration for index management and
 column type update

- Simplified the index drop and creation logic in the `upgrade` and `downgrade` functions by removing the unnecessary existence check for the `carta_porte_codes` table.
- Updated the `iva_factor` column type in the `invoice_financials` table from VARCHAR to Numeric, enhancing data integrity.
- Improved the overall clarity and efficiency of the migration script.
---
 .../versions/bccb7f8986c7_iva_factor.py       | 42 ++-----------------
 backend/tests/fixtures/builders.py            |  3 ++
 2 files changed, 7 insertions(+), 38 deletions(-)

diff --git a/backend/alembic/versions/bccb7f8986c7_iva_factor.py b/backend/alembic/versions/bccb7f8986c7_iva_factor.py
index 5a9d0d34..2c31ffb5 100644
--- a/backend/alembic/versions/bccb7f8986c7_iva_factor.py
+++ b/backend/alembic/versions/bccb7f8986c7_iva_factor.py
@@ -18,31 +18,11 @@ branch_labels: Union[str, Sequence[str], None] = None
 depends_on: Union[str, Sequence[str], None] = None
 
 
-def _public_table_exists(bind, name: str) -> bool:
-    """Evita CREATE INDEX si la tabla no existe (BD parcial / orden atípico de migraciones)."""
-    return name in sa.inspect(bind).get_table_names(schema="public")
-
-
 def upgrade() -> None:
     """Upgrade schema."""
     # ### commands auto generated by Alembic - please adjust! ###
-    bind = op.get_bind()
-    if _public_table_exists(bind, "carta_porte_codes"):
-        # Coincide con 4ad64605fad2 (schema=public). if_exists: sin índice previo.
-        op.drop_index(
-            op.f("ix_public_carta_porte_code"),
-            table_name="carta_porte_codes",
-            schema="public",
-            if_exists=True,
-        )
-        op.create_index(
-            op.f("ix_public_carta_porte_codes_code"),
-            "carta_porte_codes",
-            ["code"],
-            unique=False,
-            schema="public",
-            if_not_exists=True,
-        )
+    op.drop_index(op.f('ix_public_carta_porte_code'), table_name='carta_porte_codes')
+    op.create_index(op.f('ix_public_carta_porte_codes_code'), 'carta_porte_codes', ['code'], unique=False, schema='public')
     op.alter_column('invoice_financials', 'iva_factor',
                existing_type=sa.VARCHAR(length=10),
                type_=sa.Numeric(precision=23, scale=8),
@@ -60,20 +40,6 @@ def downgrade() -> None:
                type_=sa.VARCHAR(length=10),
                existing_nullable=True,
                schema='a76')
-    bind = op.get_bind()
-    if _public_table_exists(bind, "carta_porte_codes"):
-        op.drop_index(
-            op.f("ix_public_carta_porte_codes_code"),
-            table_name="carta_porte_codes",
-            schema="public",
-            if_exists=True,
-        )
-        op.create_index(
-            op.f("ix_public_carta_porte_code"),
-            "carta_porte_codes",
-            ["code"],
-            unique=False,
-            schema="public",
-            if_not_exists=True,
-        )
+    op.drop_index(op.f('ix_public_carta_porte_codes_code'), table_name='carta_porte_codes', schema='public')
+    op.create_index(op.f('ix_public_carta_porte_code'), 'carta_porte_codes', ['code'], unique=False)
     # ### end Alembic commands ###
diff --git a/backend/tests/fixtures/builders.py b/backend/tests/fixtures/builders.py
index 1e836dd0..6486d36b 100644
--- a/backend/tests/fixtures/builders.py
+++ b/backend/tests/fixtures/builders.py
@@ -102,6 +102,9 @@ def ensure_tenant_company(db: Session, tenant_id: int, company_id: int) -> Compa
             is_active=True,
         )
         db.add(tenant)
+        # El FK de a76.company → core.tenants exige que el tenant exista en esta transacción
+        # antes del INSERT de company; un solo flush al final puede ordenar mal (Tenant híbrido Column+Mapped).
+        db.flush()
 
     company = db.get(Company, company_id)
     if company is None:

From a6c5abbe4c0269a5f7cec782c03bc18836a5f367 Mon Sep 17 00:00:00 2001
From: AlexeerCT <acazares@aduanasoft.com.mx>
Date: Tue, 24 Mar 2026 18:52:53 -0500
Subject: [PATCH 10/10] Enhance invoice validation logic to support existing
 invoices

- Updated the `validate_common` function to accept an optional `existing_invoice` parameter, allowing for more robust validation when updating invoices.
- Introduced a new helper function, `_normalize_invoice_currency_value`, to standardize currency code handling.
- Adjusted currency validation logic to ensure proper handling of existing invoice data, preventing changes to currency types when associated line items exist.
- Modified the `validate_update` functions in both imports and exports to pass the `existing_invoice` parameter, ensuring consistent validation across different update scenarios.
---
 .../a76/invoices/common/common_validators.py  | 151 +++++++++++-------
 .../a76/invoices/exports/validators/update.py |   4 +-
 .../a76/invoices/imports/validators/update.py |   4 +-
 3 files changed, 99 insertions(+), 60 deletions(-)

diff --git a/backend/api/v1/modules/a76/invoices/common/common_validators.py b/backend/api/v1/modules/a76/invoices/common/common_validators.py
index 70f7bb5f..3f7dc48b 100644
--- a/backend/api/v1/modules/a76/invoices/common/common_validators.py
+++ b/backend/api/v1/modules/a76/invoices/common/common_validators.py
@@ -1,4 +1,4 @@
-from typing import Optional
+from typing import Any, Dict, Optional, Union
 from core.exceptions import ErrorCollector
 from sqlalchemy import func
 from sqlalchemy.orm import Session
@@ -17,8 +17,14 @@ from api.v1.modules.a76.manifests.manifest.models import Manifest
 from api.v1.modules.public.reference_data.incoterms.models import Incoterm
 from api.v1.modules.public.reference_data.currency_types.models import CurrencyType
 from api.v1.modules.public.reference_data.customs_sections.models import CustomsSection
-from core.exceptions import ErrorCollector
-from typing import Dict, Any
+
+
+def _normalize_invoice_currency_value(value) -> str:
+    """Lowercase currency code (foreign/local/manual) for comparisons."""
+    if value is None or value == "":
+        return ""
+    raw = getattr(value, "value", value)
+    return str(raw).lower()
 
 
 def invoice_exists(
@@ -174,10 +180,11 @@ def validate_required_fields_by_operation(
 
 def validate_common(
     db: Session,
-    invoice: schemas.InvoiceHeaderUpdate,
+    invoice: Union[schemas.InvoiceHeaderCreate, schemas.InvoiceHeaderUpdate],
     tenant_id: int,
     company_id: int,
     errors: ErrorCollector,
+    existing_invoice: Optional[models.InvoiceHeader] = None,
 ):
     if invoice.compliance_mx.pedimento_id:
         pedimento = (
@@ -560,69 +567,97 @@ def validate_common(
                         value=invoice.logistics.transport_num,
                     )
 
-    invoice.financials.currency = invoice.financials.currency or "foreign"
-
-    if invoice.financials.currency not in [c.value for c in Currency]:
-        errors.add_error(
-            field="financials.currency",
-            message="La Moneda proporcionada no es válida.",
-            solution=[f"Selecciona una Moneda válida: {[c.value for c in Currency]}"],
-            code="INVALID_CURRENCY",
-            value=invoice.financials.currency,
-        )
-    else:
-        # Only check for existing items during update operations (when invoice has an id)
-        if hasattr(invoice, "id"):
-            has_items = (
-                db.query(LineItem)
-                .filter(
-                    LineItem.invoice_id == invoice.id,
-                    LineItem.tenant_id == tenant_id,
-                    LineItem.company_id == company_id,
-                )
-                .first()
+    if invoice.financials:
+        submitted = invoice.financials.currency
+        stored_str = ""
+        if existing_invoice and existing_invoice.financials is not None:
+            stored_str = _normalize_invoice_currency_value(
+                existing_invoice.financials.currency
             )
-            if has_items:
-                errors.add_error(
-                    field="items",
-                    message=f"La opcion tipo de moneda {invoice.financials.currency} no puede ser modificada ya que la factura tiene items asociados.",
-                    solution=[
-                        "Verifica la moneda de los items asociados a la factura."
-                    ],
-                    code="CURRENCY_CANNOT_BE_CHANGED",
-                    value=invoice.financials.currency,
-                )
 
-    if invoice.financials.currency == "foreign":
-        invoice.financials.currency_type = "USD"
-    elif invoice.financials.currency == "local":
-        invoice.financials.currency_type = "MXN"
-    elif invoice.financials.currency == "manual":
-        if not invoice.financials.currency_type:
+        if submitted is None or submitted == "":
+            if stored_str:
+                invoice.financials.currency = stored_str
+            else:
+                invoice.financials.currency = "foreign"
+        else:
+            invoice.financials.currency = (
+                _normalize_invoice_currency_value(submitted) or "foreign"
+            )
+
+        if invoice.financials.currency not in [c.value for c in Currency]:
             errors.add_error(
-                field="financials.currency_type",
-                message="El Tipo de Moneda es obligatorio cuando la Moneda es 'manual'.",
-                solution=["Proporciona un Tipo de Moneda válido"],
-                code="REQUIRED_FIELD",
-                value=invoice.financials.currency_type,
+                field="financials.currency",
+                message="La Moneda proporcionada no es válida.",
+                solution=[f"Selecciona una Moneda válida: {[c.value for c in Currency]}"],
+                code="INVALID_CURRENCY",
+                value=invoice.financials.currency,
             )
         else:
-            currency_exists = (
-                db.query(CurrencyType)
-                .filter(CurrencyType.code == invoice.financials.currency_type)
-                .first()
-            )
-            if not currency_exists:
+            invoice_id = getattr(invoice, "id", None)
+            if (
+                invoice_id is not None
+                and existing_invoice is not None
+                and existing_invoice.financials is not None
+            ):
+                old_c = _normalize_invoice_currency_value(
+                    existing_invoice.financials.currency
+                )
+                new_c = invoice.financials.currency
+                if old_c != new_c:
+                    has_items = (
+                        db.query(LineItem)
+                        .filter(
+                            LineItem.invoice_id == invoice_id,
+                            LineItem.tenant_id == tenant_id,
+                            LineItem.company_id == company_id,
+                        )
+                        .first()
+                    )
+                    if has_items:
+                        errors.add_error(
+                            field="items",
+                            message=(
+                                "La opción tipo de moneda no puede ser modificada "
+                                "ya que la factura tiene partidas asociadas."
+                            ),
+                            solution=[
+                                "Verifica la moneda de las partidas asociadas a la factura."
+                            ],
+                            code="CURRENCY_CANNOT_BE_CHANGED",
+                            value=invoice.financials.currency,
+                        )
+
+        if invoice.financials.currency == "foreign":
+            invoice.financials.currency_type = "USD"
+        elif invoice.financials.currency == "local":
+            invoice.financials.currency_type = "MXN"
+        elif invoice.financials.currency == "manual":
+            if not invoice.financials.currency_type:
                 errors.add_error(
                     field="financials.currency_type",
-                    message="El Tipo de Moneda no existe en el Catálogo de Tipos de Moneda.",
-                    solution=[
-                        "Verifica el código del Tipo de Moneda",
-                        "Revisa el catálogo",
-                    ],
-                    code="NOT_FOUND",
+                    message="El Tipo de Moneda es obligatorio cuando la Moneda es 'manual'.",
+                    solution=["Proporciona un Tipo de Moneda válido"],
+                    code="REQUIRED_FIELD",
                     value=invoice.financials.currency_type,
                 )
+            else:
+                currency_exists = (
+                    db.query(CurrencyType)
+                    .filter(CurrencyType.code == invoice.financials.currency_type)
+                    .first()
+                )
+                if not currency_exists:
+                    errors.add_error(
+                        field="financials.currency_type",
+                        message="El Tipo de Moneda no existe en el Catálogo de Tipos de Moneda.",
+                        solution=[
+                            "Verifica el código del Tipo de Moneda",
+                            "Revisa el catálogo",
+                        ],
+                        code="NOT_FOUND",
+                        value=invoice.financials.currency_type,
+                    )
 
     if invoice.logistics.incoterm:
         incoterm_exists = (
diff --git a/backend/api/v1/modules/a76/invoices/exports/validators/update.py b/backend/api/v1/modules/a76/invoices/exports/validators/update.py
index 464c04dc..7c4b5cd4 100644
--- a/backend/api/v1/modules/a76/invoices/exports/validators/update.py
+++ b/backend/api/v1/modules/a76/invoices/exports/validators/update.py
@@ -55,7 +55,9 @@ def validate_update(
     )
     
     # Primero ejecutar validaciones comunes
-    validate_common(db, invoice, tenant_id, company_id, errors)
+    validate_common(
+        db, invoice, tenant_id, company_id, errors, existing_invoice=existing_invoice
+    )
 
     # Mapeo de columnas CSV a campos de la factura
     # Siguiendo la lógica del código Clarion original
diff --git a/backend/api/v1/modules/a76/invoices/imports/validators/update.py b/backend/api/v1/modules/a76/invoices/imports/validators/update.py
index 606df18a..565d3722 100644
--- a/backend/api/v1/modules/a76/invoices/imports/validators/update.py
+++ b/backend/api/v1/modules/a76/invoices/imports/validators/update.py
@@ -55,7 +55,9 @@ def validate_update(
     )
     
     # Primero ejecutar validaciones comunes
-    validate_common(db, invoice, tenant_id, company_id, errors)
+    validate_common(
+        db, invoice, tenant_id, company_id, errors, existing_invoice=existing_invoice
+    )
 
     # Mapeo de columnas CSV a campos de la factura
     # Siguiendo la lógica del código Clarion original