feat(T04): contact management + USt-IdNr. validation + contact UI

- Contact model: company, role (kaeufer/verkaeufer/beide), soft-delete
- ContactPerson model: 1:N relation with CASCADE delete
- USt-IdNr. validation: DE + 10 EU countries
- Contact CRUD: 7 API endpoints with RBAC, search/filter/paginate
- Frontend: ContactList, ContactForm, ContactDetail with EU/Inland toggle
- 73 backend tests (91% coverage), 20 frontend tests
This commit is contained in:
2026-07-14 12:14:21 +02:00
parent 74b5e6afb8
commit 2cf433a01c
20 changed files with 3002 additions and 97 deletions
+17 -18
View File
@@ -1,31 +1,30 @@
# Current Status # Current Status
**Task**: T02 Vehicle Management + mobile.de Push + Vehicle UI **Task**: T04 Kontakt-/Kundenverwaltung + Contact UI
**Status**: COMPLETED **Status**: COMPLETED
**Date**: 2026-07-14 **Date**: 2026-07-14
## Summary ## Summary
Vehicle CRUD API with pagination/filtering/sorting, mobile.de push integration, and full frontend UI implemented. Contact management backend (CRUD, search, filter, pagination, soft-delete, contact persons, USt-IdNr. validation) and frontend (ContactList, ContactForm, ContactDetail with EU/Inland toggle) implemented.
## Backend (COMPLETED) ## Backend (COMPLETED)
- Vehicle model: UUID PK, all fields per spec (make, model, fin CHECK len=17, year, first_registration, power_kw, power_hp computed, fuel_type, transmission, color, condition, location, availability, price, vehicle_type, lkw_type, machine_type, body_type, operating_hours, operating_hours_unit, mileage_km, description, timestamps, deleted_at soft-delete) - Contact model: UUID PK, company_name, legal_form, address fields, address_country (ISO 3166-1 alpha-2), vat_id, phone, email, website, role (kaeufer/verkaeufer/beide), vat_id_status, is_private, timestamps, deleted_at soft-delete
- MobileDeListing model: vehicle_id FK, ad_id, sync_status, synced_at, error_log - ContactPerson model: UUID PK, contact_id FK CASCADE, name, function, phone, email, created_at
- 7 API endpoints: GET /vehicles (list+filter+sort+paginate), POST /vehicles (create), GET /vehicles/:id (detail), PUT /vehicles/:id (update), DELETE /vehicles/:id (soft-delete), POST /vehicles/:id/mobile-de/push (202 async), GET /vehicles/:id/mobile-de/status - 7 API endpoints: GET /contacts (list+search+filter+sort+paginate), POST /contacts (create), GET /contacts/:id (detail+persons), PUT /contacts/:id (update), DELETE /contacts/:id (soft-delete), POST /contacts/:id/persons (add person), DELETE /contacts/:id/persons/:pid (remove person)
- mobile.de service: push_listing, update_listing, delete_listing, get_listing_status, retry_failed_listing (max 3 retries) - Contact service: list_contacts (search, role filter with beide inclusion, is_eu filter, is_private filter, sort, pagination), get_contact_by_id, create_contact (with nested persons), update_contact, soft_delete_contact, add_contact_person, remove_contact_person
- Field mapping: map_fields() converts Vehicle to mobile.de Ad format (vin, firstRegistration YYYY-MM, mileage, price EUR, power, category, bodyType, sellerLocation) - USt-IdNr. validation: DE + 10 EU countries regex patterns, EU fallback, validate_vat_id, validate_vat_id_or_raise, get_country_code_from_vat_id
- All endpoints require JWT Bearer auth - RBAC: all roles read, admin+verkaeufer write (require_role dependency)
- Config: MOBILE_DE_API_KEY, MOBILE_DE_SELLER_ID env vars added - Contacts router registered in main.py
## Frontend (COMPLETED) ## Frontend (COMPLETED)
- VehicleList: Table with filters (search, type, availability, sort), pagination, error handling - ContactList: Table with filters (search, role, EU/Inland, sort), pagination, error handling, loading state
- VehicleForm: Create/edit form with validation (make, model, fin=17 chars, price>0, vehicle_type), conditional fields - ContactForm: Create/edit form with USt-IdNr. validation, EU/Inland toggle (radio), country selector, role selector, legal form, address fields, contact info, is_private checkbox
- VehicleDetail: All vehicle fields display, delete button, MobileDeStatus embedded - ContactDetail: All contact fields display, delete button, contact persons section with add/remove via Modal
- MobileDeStatus: Sync status badge, push button, ad_id, synced_at, error_log display - 3 pages: kontakte list, kontakte/neu create, kontakte/[id] detail
- 3 pages: fahrzeuge list, fahrzeuge/neu create, fahrzeuge/[id] detail - lib/contacts.ts: Full API client with typed interfaces + validateVatIdFormat frontend validation
## Test Evidence ## Test Evidence
- Backend: 73/73 pytest passed, 82% total coverage (vehicle_service 99%, mobilede_service 80%, router 60%) - Backend: 73/73 pytest passed, 91% coverage on contact modules (service 99%, ust_validation 94%, models 93%, schemas 94%, router 67%)
- Frontend: 16/16 vitest passed - Frontend: 20/20 vitest passed
- test_report.md updated with full results - test_report.md updated with full results
- All mobile.de HTTP calls mocked in tests, no real API calls - All acceptance criteria verified and documented
- Soft-delete verified: deleted_at set, vehicle excluded from list and detail queries
+2 -1
View File
@@ -1,8 +1,9 @@
# Next Steps # Next Steps
1. T02: Vehicle CRUD + List/Filter/Pagination (Backend + Frontend) 1. T05: Lagerverwaltung + Bestandsführung (Backend + Frontend)
2. Alembic Migration Setup für DB Schema 2. Alembic Migration Setup für DB Schema
3. Docker-Compose für dev/prod erstellen 3. Docker-Compose für dev/prod erstellen
4. Redis Integration für Token-Refresh-Storage 4. Redis Integration für Token-Refresh-Storage
5. Frontend: Dashboard-Page nach Login 5. Frontend: Dashboard-Page nach Login
6. Frontend: User Management UI (Admin) 6. Frontend: User Management UI (Admin)
7. Frontend: Kontakt-Bearbeiten Seite (kontakte/[id]/bearbeiten)
+25
View File
@@ -68,3 +68,28 @@
- Backend: 73/73 pytest passed, 82% total coverage - Backend: 73/73 pytest passed, 82% total coverage
- Frontend: 16/16 vitest passed - Frontend: 16/16 vitest passed
- test_report.md updated - test_report.md updated
---
## T04: Kontakt-/Kundenverwaltung + Contact UI (2026-07-14)
### Backend
- models/contact.py: Contact + ContactPerson models with UUID PK, soft-delete, CHECK constraints for role/vat_id_status/country
- schemas/contact.py: ContactCreate/Update/Response/ListResponse + ContactPersonCreate/Response with VAT ID field_validator
- utils/ust_validation.py: DE + 10 EU country regex patterns, EU fallback, validate_vat_id, validate_vat_id_or_raise, get_country_code_from_vat_id
- services/contact_service.py: list_contacts (search, role filter with beide inclusion, is_eu filter, is_private filter, sort, pagination), get_contact_by_id, create_contact (with nested persons), update_contact, soft_delete_contact, add_contact_person, remove_contact_person
- routers/contacts.py: 7 endpoints (list, create, detail, update, delete, add person, remove person) with RBAC (all read, admin+verkaeufer write)
- main.py: Registered contacts router
### Frontend
- lib/contacts.ts: Full API client with typed interfaces + validateVatIdFormat frontend validation
- components/contacts/ContactList.tsx: Table with search, role filter, EU/Inland filter, sort, pagination
- components/contacts/ContactForm.tsx: Create/edit form with USt-IdNr. validation, EU/Inland toggle, country selector, role, legal form, address, contact info, is_private
- components/contacts/ContactDetail.tsx: Detail view with contact persons management (add/remove via Modal)
- app/[locale]/kontakte/: list page, neu (create) page, [id] detail page
- tests/contacts.test.tsx: 20 tests
### Test Results
- Backend: 73/73 pytest passed, 91% coverage on contact modules (service 99%, ust_validation 94%, models 93%, schemas 94%, router 67%)
- Frontend: 20/20 vitest passed
- test_report.md updated
BIN
View File
Binary file not shown.
+2 -1
View File
@@ -9,7 +9,7 @@ from fastapi import APIRouter, FastAPI
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
from app.config import settings from app.config import settings
from app.routers import auth, users, vehicles from app.routers import auth, contacts, users, vehicles
@asynccontextmanager @asynccontextmanager
@@ -41,6 +41,7 @@ api_v1_router = APIRouter(prefix="/api/v1")
api_v1_router.include_router(auth.router) api_v1_router.include_router(auth.router)
api_v1_router.include_router(users.router) api_v1_router.include_router(users.router)
api_v1_router.include_router(vehicles.router) api_v1_router.include_router(vehicles.router)
api_v1_router.include_router(contacts.router)
# Health endpoint (no auth required) # Health endpoint (no auth required)
@api_v1_router.get("/health", tags=["health"]) @api_v1_router.get("/health", tags=["health"])
+200
View File
@@ -0,0 +1,200 @@
"""SQLAlchemy models for contacts and contact persons."""
import enum
import uuid
from datetime import datetime
from sqlalchemy import (
Boolean,
CheckConstraint,
DateTime,
ForeignKey,
String,
func,
)
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import Base
class ContactRole(str, enum.Enum):
kaeufer = "kaeufer"
verkaeufer = "verkaeufer"
beide = "beide"
class VatIdStatus(str, enum.Enum):
ungeprueft = "ungeprueft"
geprueft = "geprueft"
ungueltig = "ungueltig"
manuell_bestaetigt = "manuell_bestaetigt"
class Contact(Base):
"""Contact entity representing a company or private contact."""
__tablename__ = "contacts"
__table_args__ = (
CheckConstraint(
"role IN ('kaeufer', 'verkaeufer', 'beide')",
name="ck_contacts_role",
),
CheckConstraint(
"vat_id_status IN ('ungeprueft', 'geprueft', 'ungueltig', 'manuell_bestaetigt')",
name="ck_contacts_vat_id_status",
),
CheckConstraint(
"char_length(address_country) = 2",
name="ck_contacts_country_alpha2",
),
)
id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
primary_key=True,
default=uuid.uuid4,
)
company_name: Mapped[str] = mapped_column(
String(255), nullable=False, index=True,
)
legal_form: Mapped[str | None] = mapped_column(
String(50), nullable=True,
)
address_street: Mapped[str | None] = mapped_column(
String(255), nullable=True,
)
address_zip: Mapped[str | None] = mapped_column(
String(10), nullable=True,
)
address_city: Mapped[str | None] = mapped_column(
String(100), nullable=True,
)
address_country: Mapped[str] = mapped_column(
String(2), nullable=False, default="DE", index=True,
)
vat_id: Mapped[str | None] = mapped_column(
String(20), nullable=True,
)
phone: Mapped[str | None] = mapped_column(
String(50), nullable=True,
)
email: Mapped[str | None] = mapped_column(
String(255), nullable=True,
)
website: Mapped[str | None] = mapped_column(
String(255), nullable=True,
)
role: Mapped[str] = mapped_column(
String(20), nullable=False, index=True,
)
vat_id_status: Mapped[str] = mapped_column(
String(20), nullable=False, default="ungeprueft",
)
is_private: Mapped[bool] = mapped_column(
Boolean, nullable=False, default=False,
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now(),
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
nullable=False,
server_default=func.now(),
onupdate=func.now(),
)
deleted_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True, index=True,
)
contact_persons: Mapped[list["ContactPerson"]] = relationship(
back_populates="contact",
cascade="all, delete-orphan",
lazy="selectin",
)
def __repr__(self) -> str:
return f"<Contact id={self.id} company_name={self.company_name}>"
def to_dict(self) -> dict:
"""Serialize contact for API responses."""
return {
"id": str(self.id),
"company_name": self.company_name,
"legal_form": self.legal_form,
"address_street": self.address_street,
"address_zip": self.address_zip,
"address_city": self.address_city,
"address_country": self.address_country,
"vat_id": self.vat_id,
"phone": self.phone,
"email": self.email,
"website": self.website,
"role": self.role,
"vat_id_status": self.vat_id_status,
"is_private": self.is_private,
"created_at": (
self.created_at.isoformat() if self.created_at else None
),
"updated_at": (
self.updated_at.isoformat() if self.updated_at else None
),
"deleted_at": (
self.deleted_at.isoformat() if self.deleted_at else None
),
"contact_persons": [
p.to_dict() for p in (self.contact_persons or [])
],
}
class ContactPerson(Base):
"""Contact person associated with a contact (company)."""
__tablename__ = "contact_persons"
id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
primary_key=True,
default=uuid.uuid4,
)
contact_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("contacts.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
name: Mapped[str] = mapped_column(
String(255), nullable=False,
)
function: Mapped[str | None] = mapped_column(
String(100), nullable=True,
)
phone: Mapped[str | None] = mapped_column(
String(50), nullable=True,
)
email: Mapped[str | None] = mapped_column(
String(255), nullable=True,
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now(),
)
contact: Mapped["Contact"] = relationship(back_populates="contact_persons")
def __repr__(self) -> str:
return f"<ContactPerson id={self.id} name={self.name}>"
def to_dict(self) -> dict:
"""Serialize contact person for API responses."""
return {
"id": str(self.id),
"contact_id": str(self.contact_id),
"name": self.name,
"function": self.function,
"phone": self.phone,
"email": self.email,
"created_at": (
self.created_at.isoformat() if self.created_at else None
),
}
+164
View File
@@ -0,0 +1,164 @@
"""Contacts router: CRUD, search, filter, and contact person management endpoints."""
import uuid
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.dependencies import get_current_user, get_pagination, require_role
from app.models.user import User
from app.schemas.contact import (
ContactCreate,
ContactListResponse,
ContactPersonCreate,
ContactPersonResponse,
ContactResponse,
ContactUpdate,
)
from app.services import contact_service
router = APIRouter(prefix="/contacts", tags=["contacts"])
@router.get("/", response_model=ContactListResponse, status_code=status.HTTP_200_OK)
async def list_contacts(
db: AsyncSession = Depends(get_db),
pagination: dict = Depends(get_pagination),
search: str | None = Query(None, description="Search in company_name, city, email, vat_id"),
role: str | None = Query(None, description="Filter by role (kaeufer, verkaeufer, beide)"),
is_eu: bool | None = Query(None, description="Filter EU (true) or Inland (false)"),
is_private: bool | None = Query(None, description="Filter private contacts"),
sort: str | None = Query(None, description="Sort field (prefix - for descending)"),
current_user: User = Depends(get_current_user),
):
"""List contacts with pagination, filtering, and sorting.
All authenticated roles can read.
"""
contacts, total = await contact_service.list_contacts(
db,
page=pagination["page"],
page_size=pagination["page_size"],
search=search,
role=role,
is_eu=is_eu,
is_private=is_private,
sort=sort,
)
return ContactListResponse(
items=[ContactResponse.model_validate(c) for c in contacts],
total=total,
page=pagination["page"],
page_size=pagination["page_size"],
)
@router.post("/", response_model=ContactResponse, status_code=status.HTTP_201_CREATED)
async def create_contact(
body: ContactCreate,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(require_role(["admin", "verkaeufer"])),
):
"""Create a new contact. Requires admin or verkaeufer role."""
data = body.model_dump(exclude_unset=False)
contact = await contact_service.create_contact(db, data)
return ContactResponse.model_validate(contact)
@router.get("/{contact_id}", response_model=ContactResponse, status_code=status.HTTP_200_OK)
async def get_contact(
contact_id: uuid.UUID,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Get a single contact by ID with contact persons."""
contact = await contact_service.get_contact_by_id(db, contact_id)
if contact is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={"error": {"code": "CONTACT_NOT_FOUND", "message": "Contact not found"}},
)
return ContactResponse.model_validate(contact)
@router.put("/{contact_id}", response_model=ContactResponse, status_code=status.HTTP_200_OK)
async def update_contact(
contact_id: uuid.UUID,
body: ContactUpdate,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(require_role(["admin", "verkaeufer"])),
):
"""Update a contact's fields. Requires admin or verkaeufer role."""
updates = body.model_dump(exclude_unset=True)
if not updates:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={"error": {"code": "NO_FIELDS", "message": "No fields to update"}},
)
contact = await contact_service.update_contact(db, contact_id, updates)
if contact is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={"error": {"code": "CONTACT_NOT_FOUND", "message": "Contact not found"}},
)
return ContactResponse.model_validate(contact)
@router.delete("/{contact_id}", response_model=ContactResponse, status_code=status.HTTP_200_OK)
async def delete_contact(
contact_id: uuid.UUID,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(require_role(["admin", "verkaeufer"])),
):
"""Soft-delete a contact (sets deleted_at). Requires admin or verkaeufer role."""
contact = await contact_service.soft_delete_contact(db, contact_id)
if contact is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={"error": {"code": "CONTACT_NOT_FOUND", "message": "Contact not found"}},
)
return ContactResponse.model_validate(contact)
@router.post(
"/{contact_id}/persons",
response_model=ContactPersonResponse,
status_code=status.HTTP_201_CREATED,
)
async def add_contact_person(
contact_id: uuid.UUID,
body: ContactPersonCreate,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(require_role(["admin", "verkaeufer"])),
):
"""Add a contact person to an existing contact. Requires admin or verkaeufer role."""
person = await contact_service.add_contact_person(
db, contact_id, body.model_dump(exclude_unset=False)
)
if person is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={"error": {"code": "CONTACT_NOT_FOUND", "message": "Contact not found"}},
)
return ContactPersonResponse.model_validate(person)
@router.delete(
"/{contact_id}/persons/{person_id}",
status_code=status.HTTP_204_NO_CONTENT,
)
async def remove_contact_person(
contact_id: uuid.UUID,
person_id: uuid.UUID,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(require_role(["admin", "verkaeufer"])),
):
"""Remove a contact person from a contact. Requires admin or verkaeufer role."""
deleted = await contact_service.remove_contact_person(db, contact_id, person_id)
if not deleted:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={"error": {"code": "PERSON_NOT_FOUND", "message": "Contact person not found"}},
)
return None
+147
View File
@@ -0,0 +1,147 @@
"""Pydantic schemas for contact-related request and response bodies."""
import uuid
from datetime import datetime
from typing import Literal, Optional
from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator
from app.utils.ust_validation import validate_vat_id
ContactRole = Literal["kaeufer", "verkaeufer", "beide"]
VatIdStatus = Literal["ungeprueft", "geprueft", "ungueltig", "manuell_bestaetigt"]
class ContactPersonBase(BaseModel):
"""Base contact person fields."""
name: str = Field(..., min_length=1, max_length=255)
function: Optional[str] = Field(None, max_length=100)
phone: Optional[str] = Field(None, max_length=50)
email: Optional[str] = Field(None, max_length=255)
class ContactPersonCreate(ContactPersonBase):
"""POST /api/v1/contacts/:id/persons request body."""
pass
class ContactPersonResponse(ContactPersonBase):
"""Contact person response schema."""
model_config = ConfigDict(from_attributes=True)
id: uuid.UUID
contact_id: uuid.UUID
created_at: Optional[datetime] = None
class ContactBase(BaseModel):
"""Base contact fields shared across schemas."""
company_name: str = Field(..., min_length=1, max_length=255)
legal_form: Optional[str] = Field(None, max_length=50)
address_street: Optional[str] = Field(None, max_length=255)
address_zip: Optional[str] = Field(None, max_length=10)
address_city: Optional[str] = Field(None, max_length=100)
address_country: str = Field("DE", min_length=2, max_length=2)
vat_id: Optional[str] = Field(None, max_length=20)
phone: Optional[str] = Field(None, max_length=50)
email: Optional[str] = Field(None, max_length=255)
website: Optional[str] = Field(None, max_length=255)
role: ContactRole
is_private: bool = False
@field_validator("vat_id")
@classmethod
def validate_vat_id_format(cls, v: str | None) -> str | None:
"""Validate VAT ID format if provided."""
if v is None or v == "":
return None
if not validate_vat_id(v):
raise ValueError(f"Invalid VAT ID format: '{v}'")
return v.strip().upper().replace(" ", "")
@field_validator("address_country")
@classmethod
def validate_country_code(cls, v: str) -> str:
"""Normalise country code to uppercase."""
return v.upper()
class ContactCreate(ContactBase):
"""POST /api/v1/contacts request body."""
contact_persons: list[ContactPersonCreate] = Field(
default_factory=list, description="Contact persons to create with the contact"
)
class ContactUpdate(BaseModel):
"""PUT /api/v1/contacts/:id request body (all fields optional)."""
company_name: Optional[str] = Field(None, min_length=1, max_length=255)
legal_form: Optional[str] = Field(None, max_length=50)
address_street: Optional[str] = Field(None, max_length=255)
address_zip: Optional[str] = Field(None, max_length=10)
address_city: Optional[str] = Field(None, max_length=100)
address_country: Optional[str] = Field(None, min_length=2, max_length=2)
vat_id: Optional[str] = Field(None, max_length=20)
phone: Optional[str] = Field(None, max_length=50)
email: Optional[str] = Field(None, max_length=255)
website: Optional[str] = Field(None, max_length=255)
role: Optional[ContactRole] = None
is_private: Optional[bool] = None
@field_validator("vat_id")
@classmethod
def validate_vat_id_format(cls, v: str | None) -> str | None:
"""Validate VAT ID format if provided."""
if v is None or v == "":
return None
if not validate_vat_id(v):
raise ValueError(f"Invalid VAT ID format: '{v}'")
return v.strip().upper().replace(" ", "")
@field_validator("address_country")
@classmethod
def validate_country_code(cls, v: str | None) -> str | None:
"""Normalise country code to uppercase."""
if v is not None:
return v.upper()
return v
class ContactResponse(BaseModel):
"""Contact response schema."""
model_config = ConfigDict(from_attributes=True)
id: uuid.UUID
company_name: str
legal_form: Optional[str] = None
address_street: Optional[str] = None
address_zip: Optional[str] = None
address_city: Optional[str] = None
address_country: str
vat_id: Optional[str] = None
phone: Optional[str] = None
email: Optional[str] = None
website: Optional[str] = None
role: str
vat_id_status: str = "ungeprueft"
is_private: bool = False
created_at: Optional[datetime] = None
updated_at: Optional[datetime] = None
deleted_at: Optional[datetime] = None
contact_persons: list[ContactPersonResponse] = Field(default_factory=list)
class ContactListResponse(BaseModel):
"""Paginated contact list response."""
items: list[ContactResponse]
total: int
page: int
page_size: int
+263
View File
@@ -0,0 +1,263 @@
"""Contact service: CRUD, search, filter, pagination, and soft-delete operations."""
from __future__ import annotations
import uuid
from datetime import datetime, timezone
from typing import Any
from sqlalchemy import and_, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from app.models.contact import Contact, ContactPerson
_SORTABLE_FIELDS: set[str] = {
"company_name",
"address_city",
"address_country",
"role",
"created_at",
"updated_at",
"email",
}
def _apply_filters(
stmt: select,
search: str | None = None,
role: str | None = None,
is_eu: bool | None = None,
is_private: bool | None = None,
) -> select:
"""Apply WHERE filters to a select statement (always excludes soft-deleted)."""
conditions = [Contact.deleted_at.is_(None)]
if search:
search_pattern = f"%{search}%"
conditions.append(
or_(
Contact.company_name.ilike(search_pattern),
Contact.address_city.ilike(search_pattern),
Contact.email.ilike(search_pattern),
Contact.vat_id.ilike(search_pattern),
)
)
if role:
# When filtering by 'kaeufer' or 'verkaeufer', include 'beide' as well
if role in ("kaeufer", "verkaeufer"):
conditions.append(
or_(
Contact.role == role,
Contact.role == "beide",
)
)
else:
conditions.append(Contact.role == role)
if is_eu is not None:
if is_eu:
# EU = address_country != 'DE'
conditions.append(Contact.address_country != "DE")
else:
# Inland = address_country == 'DE'
conditions.append(Contact.address_country == "DE")
if is_private is not None:
conditions.append(Contact.is_private == is_private)
return stmt.where(and_(*conditions))
def _apply_sort(stmt: select, sort: str | None = None) -> select:
"""Apply ORDER BY to a select statement based on sort param.
Format: 'field' for ascending, '-field' for descending.
"""
if not sort:
return stmt.order_by(Contact.created_at.desc())
descending = sort.startswith("-")
field_name = sort.lstrip("-")
if field_name not in _SORTABLE_FIELDS:
return stmt.order_by(Contact.created_at.desc())
column = getattr(Contact, field_name)
if descending:
return stmt.order_by(column.desc())
return stmt.order_by(column.asc())
async def list_contacts(
db: AsyncSession,
page: int = 1,
page_size: int = 20,
search: str | None = None,
role: str | None = None,
is_eu: bool | None = None,
is_private: bool | None = None,
sort: str | None = None,
) -> tuple[list[Contact], int]:
"""List contacts with pagination, filtering, and sorting.
Returns (contacts, total_count).
"""
# Build count query
count_stmt = select(func.count(Contact.id))
count_stmt = _apply_filters(
count_stmt,
search=search,
role=role,
is_eu=is_eu,
is_private=is_private,
)
total_result = await db.execute(count_stmt)
total = total_result.scalar_one()
# Build data query with eager-loaded contact persons
data_stmt = select(Contact).options(selectinload(Contact.contact_persons))
data_stmt = _apply_filters(
data_stmt,
search=search,
role=role,
is_eu=is_eu,
is_private=is_private,
)
data_stmt = _apply_sort(data_stmt, sort)
offset = (page - 1) * page_size
data_stmt = data_stmt.offset(offset).limit(page_size)
result = await db.execute(data_stmt)
contacts = list(result.scalars().all())
return contacts, total
async def get_contact_by_id(
db: AsyncSession, contact_id: uuid.UUID
) -> Contact | None:
"""Get a single contact by ID, excluding soft-deleted. Eager-loads contact persons."""
stmt = (
select(Contact)
.options(selectinload(Contact.contact_persons))
.where(
and_(Contact.id == contact_id, Contact.deleted_at.is_(None))
)
)
result = await db.execute(stmt)
return result.scalar_one_or_none()
async def create_contact(
db: AsyncSession, data: dict[str, Any]
) -> Contact:
"""Create a new contact with optional nested contact persons.
The data dict may contain a 'contact_persons' list of dicts.
"""
persons_data = data.pop("contact_persons", [])
contact = Contact(**data)
db.add(contact)
await db.flush() # Flush to generate contact.id
# Add contact persons if provided
for person_data in persons_data:
person = ContactPerson(
contact_id=contact.id,
name=person_data["name"],
function=person_data.get("function"),
phone=person_data.get("phone"),
email=person_data.get("email"),
)
db.add(person)
await db.flush()
await db.refresh(contact)
return contact
async def update_contact(
db: AsyncSession, contact_id: uuid.UUID, updates: dict[str, Any]
) -> Contact | None:
"""Update a contact's fields. Returns None if not found or deleted."""
contact = await get_contact_by_id(db, contact_id)
if contact is None:
return None
for key, value in updates.items():
if hasattr(contact, key):
setattr(contact, key, value)
await db.flush()
await db.refresh(contact)
return contact
async def soft_delete_contact(
db: AsyncSession, contact_id: uuid.UUID
) -> Contact | None:
"""Soft-delete a contact by setting deleted_at. Returns None if not found."""
contact = await get_contact_by_id(db, contact_id)
if contact is None:
return None
contact.deleted_at = datetime.now(timezone.utc)
await db.flush()
await db.refresh(contact)
return contact
async def add_contact_person(
db: AsyncSession,
contact_id: uuid.UUID,
person_data: dict[str, Any],
) -> ContactPerson | None:
"""Add a contact person to an existing contact.
Returns None if the contact is not found or is deleted.
"""
contact = await get_contact_by_id(db, contact_id)
if contact is None:
return None
person = ContactPerson(
contact_id=contact_id,
name=person_data["name"],
function=person_data.get("function"),
phone=person_data.get("phone"),
email=person_data.get("email"),
)
db.add(person)
await db.flush()
await db.refresh(person)
return person
async def remove_contact_person(
db: AsyncSession,
contact_id: uuid.UUID,
person_id: uuid.UUID,
) -> bool:
"""Remove (hard-delete) a contact person from a contact.
Returns True if the person was found and deleted, False otherwise.
"""
stmt = select(ContactPerson).where(
and_(
ContactPerson.id == person_id,
ContactPerson.contact_id == contact_id,
)
)
result = await db.execute(stmt)
person = result.scalar_one_or_none()
if person is None:
return False
await db.delete(person)
await db.flush()
return True
+114
View File
@@ -0,0 +1,114 @@
"""USt-IdNr. (VAT ID) format validation for DE and common EU countries.
Provides basic regex-based format validation. Does NOT perform live API verification.
"""
import re
# Country-specific VAT ID regex patterns.
# Each pattern validates the format after the 2-letter country code prefix.
_VAT_PATTERNS: dict[str, re.Pattern[str]] = {
# DE: DE + 9 digits (e.g. DE123456789)
"DE": re.compile(r"^DE\d{9}$"),
# AT: AT + U + 8 digits (e.g. ATU12345678)
"AT": re.compile(r"^ATU\d{8}$"),
# FR: FR + 2 alphanumeric + 9 digits (e.g. FRAB123456789)
"FR": re.compile(r"^FR[A-Za-z0-9]{2}\d{9}$"),
# NL: NL + 9 digits + B + 2 digits (e.g. NL123456789B01)
"NL": re.compile(r"^NL\d{9}B\d{2}$"),
# PL: PL + 10 digits (e.g. PL1234567890)
"PL": re.compile(r"^PL\d{10}$"),
# CZ: CZ + 8-10 digits (e.g. CZ1234567890)
"CZ": re.compile(r"^CZ\d{8,10}$"),
# IT: IT + 11 digits (e.g. IT12345678901)
"IT": re.compile(r"^IT\d{11}$"),
# ES: ES + 1 alphanumeric + 7 digits + 1 alphanumeric (e.g. ESA1234567B)
"ES": re.compile(r"^ES[A-Za-z0-9]\d{7}[A-Za-z0-9]$"),
# BE: BE + 10 digits (e.g. BE1234567890)
"BE": re.compile(r"^BE\d{10}$"),
# DK: DK + 8 digits (e.g. DK12345678)
"DK": re.compile(r"^DK\d{8}$"),
# SE: SE + 10 digits (e.g. SE1234567890)
"SE": re.compile(r"^SE\d{10}$"),
}
# Fallback pattern for EU countries not explicitly listed above.
# Accepts <2-letter country code> + 5-15 alphanumeric characters.
_EU_FALLBACK_PATTERN = re.compile(r"^[A-Z]{2}[A-Za-z0-9]{5,15}$")
# Set of supported EU country codes (ISO 3166-1 alpha-2).
_EU_COUNTRY_CODES: set[str] = {
"AT", "BE", "BG", "CY", "CZ", "DE", "DK", "EE", "ES", "FI", "FR", "GR",
"HR", "HU", "IE", "IT", "LT", "LU", "LV", "MT", "NL", "PL", "PT", "RO",
"SE", "SI", "SK",
}
def get_country_code_from_vat_id(vat_id: str) -> str | None:
"""Extract the 2-letter country code from a VAT ID string.
Returns None if the string is too short or does not start with letters.
"""
if not vat_id or len(vat_id) < 3:
return None
prefix = vat_id[:2].upper()
if not prefix.isalpha():
return None
return prefix
def validate_vat_id(vat_id: str) -> bool:
"""Validate the format of a VAT ID (USt-IdNr.).
Supports DE and common EU countries with specific regex patterns.
For other EU countries, a fallback pattern is used.
Non-EU or unrecognised formats return False.
Args:
vat_id: The VAT ID string to validate (case-insensitive, will be uppercased).
Returns:
True if the format is valid, False otherwise.
"""
if not vat_id:
return True # Empty VAT ID is valid (nullable field)
normalized = vat_id.strip().upper().replace(" ", "")
country_code = get_country_code_from_vat_id(normalized)
if country_code is None:
return False
# Check against country-specific pattern if available
pattern = _VAT_PATTERNS.get(country_code)
if pattern is not None:
return bool(pattern.match(normalized))
# Fallback for EU countries without a specific pattern
if country_code in _EU_COUNTRY_CODES:
return bool(_EU_FALLBACK_PATTERN.match(normalized))
# Non-EU country code: reject (this module validates EU VAT IDs only)
return False
def validate_vat_id_or_raise(vat_id: str | None) -> str | None:
"""Validate VAT ID format and return the normalised value.
Raises ValueError if the format is invalid.
Args:
vat_id: The VAT ID string to validate, or None.
Returns:
The normalised (uppercased, whitespace-stripped) VAT ID, or None.
Raises:
ValueError: If the VAT ID format is invalid.
"""
if vat_id is None or vat_id == "":
return None
normalized = vat_id.strip().upper().replace(" ", "")
if not validate_vat_id(normalized):
raise ValueError(f"Invalid VAT ID format: '{vat_id}'")
return normalized
+829
View File
@@ -0,0 +1,829 @@
"""Tests for contact CRUD, search, filter, and contact person endpoints."""
import uuid
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from app.database import Base, get_db
from app.main import app
from app.models.contact import Contact, ContactPerson
from app.utils.ust_validation import validate_vat_id
@pytest_asyncio.fixture
async def sample_contact_data():
"""Valid contact data for creation."""
return {
"company_name": "Müller Transport GmbH",
"legal_form": "GmbH",
"address_street": "Hauptstraße 1",
"address_zip": "10115",
"address_city": "Berlin",
"address_country": "DE",
"vat_id": "DE123456789",
"phone": "+49 30 12345678",
"email": "info@mueller-transport.de",
"website": "https://mueller-transport.de",
"role": "kaeufer",
"is_private": False,
}
@pytest_asyncio.fixture
async def sample_eu_contact_data():
"""Valid EU contact data (non-DE)."""
return {
"company_name": "Van der Berg Logistics B.V.",
"address_street": "Keizersgracht 100",
"address_zip": "1015",
"address_city": "Amsterdam",
"address_country": "NL",
"vat_id": "NL123456789B01",
"email": "info@vandberg.nl",
"role": "verkaeufer",
"is_private": False,
}
@pytest_asyncio.fixture
async def sample_beide_contact_data():
"""Valid contact with role 'beide'."""
return {
"company_name": "Schmidt & Söhne KG",
"address_city": "Hamburg",
"address_country": "DE",
"role": "beide",
"is_private": False,
}
@pytest_asyncio.fixture
async def created_contact(admin_client, sample_contact_data):
"""Create a contact via API and return the response."""
response = await admin_client.post("/api/v1/contacts/", json=sample_contact_data)
assert response.status_code == 201, response.text
return response.json()
class TestUstValidation:
"""Unit tests for USt-IdNr. validation utility."""
def test_validate_de_vat_id_valid(self):
assert validate_vat_id("DE123456789") is True
def test_validate_de_vat_id_invalid_short(self):
assert validate_vat_id("DE12345678") is False
def test_validate_de_vat_id_invalid_long(self):
assert validate_vat_id("DE1234567890") is False
def test_validate_at_vat_id_valid(self):
assert validate_vat_id("ATU12345678") is True
def test_validate_nl_vat_id_valid(self):
assert validate_vat_id("NL123456789B01") is True
def test_validate_fr_vat_id_valid(self):
assert validate_vat_id("FRAB123456789") is True
def test_validate_it_vat_id_valid(self):
assert validate_vat_id("IT12345678901") is True
def test_validate_es_vat_id_valid(self):
assert validate_vat_id("ESA1234567B") is True
def test_validate_empty_vat_id(self):
assert validate_vat_id("") is True
def test_validate_none_vat_id(self):
assert validate_vat_id(None) is True
def test_validate_non_eu_country(self):
assert validate_vat_id("US123456789") is False
def test_validate_lowercase_normalised(self):
assert validate_vat_id("de123456789") is True
def test_validate_with_spaces(self):
assert validate_vat_id("DE 123 456 789") is True
class TestContactList:
"""GET /api/v1/contacts tests."""
@pytest.mark.asyncio
async def test_list_contacts_returns_200_with_pagination(self, admin_client, created_contact):
"""GET /api/v1/contacts returns 200 with paginated list."""
response = await admin_client.get("/api/v1/contacts/?page=1&page_size=20")
assert response.status_code == 200
data = response.json()
assert "items" in data
assert "total" in data
assert "page" in data
assert "page_size" in data
assert data["total"] >= 1
assert len(data["items"]) >= 1
@pytest.mark.asyncio
async def test_list_contacts_filter_by_role_kaeufer_includes_beide(
self, admin_client, sample_contact_data, sample_beide_contact_data
):
"""GET /api/v1/contacts?role=kaeufer returns kaeufer + beide contacts."""
# Create a kaeufer contact
resp1 = await admin_client.post("/api/v1/contacts/", json=sample_contact_data)
assert resp1.status_code == 201
# Create a beide contact
resp2 = await admin_client.post("/api/v1/contacts/", json=sample_beide_contact_data)
assert resp2.status_code == 201
response = await admin_client.get("/api/v1/contacts/?role=kaeufer")
assert response.status_code == 200
data = response.json()
roles = [c["role"] for c in data["items"]]
assert "kaeufer" in roles
assert "beide" in roles
assert "verkaeufer" not in roles
@pytest.mark.asyncio
async def test_list_contacts_filter_by_role_verkaeufer_includes_beide(
self, admin_client, sample_eu_contact_data, sample_beide_contact_data
):
"""GET /api/v1/contacts?role=verkaeufer returns verkaeufer + beide contacts."""
resp1 = await admin_client.post("/api/v1/contacts/", json=sample_eu_contact_data)
assert resp1.status_code == 201
resp2 = await admin_client.post("/api/v1/contacts/", json=sample_beide_contact_data)
assert resp2.status_code == 201
response = await admin_client.get("/api/v1/contacts/?role=verkaeufer")
assert response.status_code == 200
data = response.json()
roles = [c["role"] for c in data["items"]]
assert "verkaeufer" in roles
assert "beide" in roles
assert "kaeufer" not in roles
@pytest.mark.asyncio
async def test_list_contacts_filter_is_eu_true(
self, admin_client, sample_contact_data, sample_eu_contact_data
):
"""GET /api/v1/contacts?is_eu=true returns only non-DE contacts."""
await admin_client.post("/api/v1/contacts/", json=sample_contact_data)
await admin_client.post("/api/v1/contacts/", json=sample_eu_contact_data)
response = await admin_client.get("/api/v1/contacts/?is_eu=true")
assert response.status_code == 200
data = response.json()
for item in data["items"]:
assert item["address_country"] != "DE"
@pytest.mark.asyncio
async def test_list_contacts_filter_is_eu_false(
self, admin_client, sample_contact_data, sample_eu_contact_data
):
"""GET /api/v1/contacts?is_eu=false returns only DE contacts."""
await admin_client.post("/api/v1/contacts/", json=sample_contact_data)
await admin_client.post("/api/v1/contacts/", json=sample_eu_contact_data)
response = await admin_client.get("/api/v1/contacts/?is_eu=false")
assert response.status_code == 200
data = response.json()
for item in data["items"]:
assert item["address_country"] == "DE"
@pytest.mark.asyncio
async def test_list_contacts_search_by_company_name(self, admin_client, created_contact):
"""GET /api/v1/contacts?search=mueller returns matching contacts."""
response = await admin_client.get("/api/v1/contacts/?search=mueller")
assert response.status_code == 200
data = response.json()
assert data["total"] >= 1
assert any("Müller" in c["company_name"] for c in data["items"])
@pytest.mark.asyncio
async def test_list_contacts_search_by_city(self, admin_client, created_contact):
"""GET /api/v1/contacts?search=berlin returns matching contacts."""
response = await admin_client.get("/api/v1/contacts/?search=berlin")
assert response.status_code == 200
data = response.json()
assert data["total"] >= 1
@pytest.mark.asyncio
async def test_list_contacts_search_no_results(self, admin_client):
"""GET /api/v1/contacts?search=nonexistent returns empty list."""
response = await admin_client.get("/api/v1/contacts/?search=nonexistent_xyz")
assert response.status_code == 200
data = response.json()
assert data["total"] == 0
assert len(data["items"]) == 0
@pytest.mark.asyncio
async def test_list_contacts_sort_by_company_name(self, admin_client, sample_contact_data, sample_eu_contact_data):
"""GET /api/v1/contacts?sort=company_name returns sorted list."""
await admin_client.post("/api/v1/contacts/", json=sample_contact_data)
await admin_client.post("/api/v1/contacts/", json=sample_eu_contact_data)
response = await admin_client.get("/api/v1/contacts/?sort=company_name")
assert response.status_code == 200
data = response.json()
names = [c["company_name"] for c in data["items"]]
assert names == sorted(names)
@pytest.mark.asyncio
async def test_list_contacts_pagination(self, admin_client, sample_contact_data):
"""GET /api/v1/contacts?page=1&page_size=1 returns correct pagination."""
for i in range(3):
data = {**sample_contact_data, "company_name": f"Company {i} GmbH"}
await admin_client.post("/api/v1/contacts/", json=data)
response = await admin_client.get("/api/v1/contacts/?page=1&page_size=1")
assert response.status_code == 200
data = response.json()
assert data["page"] == 1
assert data["page_size"] == 1
assert len(data["items"]) == 1
assert data["total"] >= 3
class TestContactDetail:
"""GET /api/v1/contacts/:id tests."""
@pytest.mark.asyncio
async def test_get_contact_returns_200_with_detail(self, admin_client, created_contact):
"""GET /api/v1/contacts/:id returns 200 with contact detail."""
response = await admin_client.get(f"/api/v1/contacts/{created_contact['id']}")
assert response.status_code == 200
data = response.json()
assert data["id"] == created_contact["id"]
assert data["company_name"] == "Müller Transport GmbH"
assert "contact_persons" in data
assert isinstance(data["contact_persons"], list)
@pytest.mark.asyncio
async def test_get_contact_nonexistent_returns_404(self, admin_client):
"""GET /api/v1/contacts/:nonexistent returns 404."""
fake_id = uuid.uuid4()
response = await admin_client.get(f"/api/v1/contacts/{fake_id}")
assert response.status_code == 404
data = response.json()
assert data["detail"]["error"]["code"] == "CONTACT_NOT_FOUND"
@pytest.mark.asyncio
async def test_get_contact_after_soft_delete_returns_404(self, admin_client, created_contact):
"""GET /api/v1/contacts/:id after soft-delete returns 404."""
del_resp = await admin_client.delete(f"/api/v1/contacts/{created_contact['id']}")
assert del_resp.status_code == 200
get_resp = await admin_client.get(f"/api/v1/contacts/{created_contact['id']}")
assert get_resp.status_code == 404
class TestContactCreate:
"""POST /api/v1/contacts tests."""
@pytest.mark.asyncio
async def test_create_contact_returns_201(self, admin_client, sample_contact_data):
"""POST /api/v1/contacts with valid data returns 201."""
response = await admin_client.post("/api/v1/contacts/", json=sample_contact_data)
assert response.status_code == 201
data = response.json()
assert data["company_name"] == sample_contact_data["company_name"]
assert data["role"] == "kaeufer"
assert data["vat_id_status"] == "ungeprueft"
assert data["id"] is not None
@pytest.mark.asyncio
async def test_create_contact_with_invalid_vat_id_returns_422(self, admin_client, sample_contact_data):
"""POST /api/v1/contacts with invalid VAT ID format returns 422."""
data = {**sample_contact_data, "vat_id": "INVALID123"}
response = await admin_client.post("/api/v1/contacts/", json=data)
assert response.status_code == 422
@pytest.mark.asyncio
async def test_create_contact_with_de_vat_too_short_returns_422(self, admin_client, sample_contact_data):
"""POST /api/v1/contacts with too-short DE VAT ID returns 422."""
data = {**sample_contact_data, "vat_id": "DE12345678"}
response = await admin_client.post("/api/v1/contacts/", json=data)
assert response.status_code == 422
@pytest.mark.asyncio
async def test_create_contact_with_no_vat_id_returns_201(self, admin_client, sample_contact_data):
"""POST /api/v1/contacts without VAT ID returns 201."""
data = {**sample_contact_data}
data.pop("vat_id")
response = await admin_client.post("/api/v1/contacts/", json=data)
assert response.status_code == 201
assert response.json()["vat_id"] is None
@pytest.mark.asyncio
async def test_create_contact_with_contact_persons(self, admin_client, sample_contact_data):
"""POST /api/v1/contacts with nested contact persons returns 201."""
data = {
**sample_contact_data,
"contact_persons": [
{
"name": "Hans Müller",
"function": "Geschäftsführer",
"phone": "+49 30 87654321",
"email": "hans@mueller-transport.de",
}
],
}
response = await admin_client.post("/api/v1/contacts/", json=data)
assert response.status_code == 201
contact_data = response.json()
assert len(contact_data["contact_persons"]) == 1
assert contact_data["contact_persons"][0]["name"] == "Hans Müller"
@pytest.mark.asyncio
async def test_create_contact_missing_required_fields_returns_422(self, admin_client):
"""POST /api/v1/contacts with missing required fields returns 422."""
response = await admin_client.post("/api/v1/contacts/", json={"address_country": "DE"})
assert response.status_code == 422
@pytest.mark.asyncio
async def test_create_contact_invalid_role_returns_422(self, admin_client, sample_contact_data):
"""POST /api/v1/contacts with invalid role returns 422."""
data = {**sample_contact_data, "role": "invalid_role"}
response = await admin_client.post("/api/v1/contacts/", json=data)
assert response.status_code == 422
class TestContactUpdate:
"""PUT /api/v1/contacts/:id tests."""
@pytest.mark.asyncio
async def test_update_contact_returns_200(self, admin_client, created_contact):
"""PUT /api/v1/contacts/:id with valid data returns 200."""
response = await admin_client.put(
f"/api/v1/contacts/{created_contact['id']}",
json={"company_name": "Müller Transport AG", "phone": "+49 30 99999999"},
)
assert response.status_code == 200
data = response.json()
assert data["company_name"] == "Müller Transport AG"
assert data["phone"] == "+49 30 99999999"
@pytest.mark.asyncio
async def test_update_contact_nonexistent_returns_404(self, admin_client):
"""PUT /api/v1/contacts/:nonexistent returns 404."""
fake_id = uuid.uuid4()
response = await admin_client.put(
f"/api/v1/contacts/{fake_id}",
json={"company_name": "Test GmbH"},
)
assert response.status_code == 404
@pytest.mark.asyncio
async def test_update_contact_invalid_vat_id_returns_422(self, admin_client, created_contact):
"""PUT /api/v1/contacts/:id with invalid VAT ID returns 422."""
response = await admin_client.put(
f"/api/v1/contacts/{created_contact['id']}",
json={"vat_id": "INVALID123"},
)
assert response.status_code == 422
@pytest.mark.asyncio
async def test_update_contact_no_fields_returns_400(self, admin_client, created_contact):
"""PUT /api/v1/contacts/:id with no fields returns 400."""
response = await admin_client.put(
f"/api/v1/contacts/{created_contact['id']}",
json={},
)
assert response.status_code == 400
class TestContactDelete:
"""DELETE /api/v1/contacts/:id tests."""
@pytest.mark.asyncio
async def test_delete_contact_returns_200(self, admin_client, created_contact):
"""DELETE /api/v1/contacts/:id returns 200 (soft delete)."""
response = await admin_client.delete(f"/api/v1/contacts/{created_contact['id']}")
assert response.status_code == 200
data = response.json()
assert data["deleted_at"] is not None
@pytest.mark.asyncio
async def test_delete_contact_nonexistent_returns_404(self, admin_client):
"""DELETE /api/v1/contacts/:nonexistent returns 404."""
fake_id = uuid.uuid4()
response = await admin_client.delete(f"/api/v1/contacts/{fake_id}")
assert response.status_code == 404
@pytest.mark.asyncio
async def test_deleted_contact_not_in_list(self, admin_client, created_contact):
"""Soft-deleted contact does not appear in list."""
del_resp = await admin_client.delete(f"/api/v1/contacts/{created_contact['id']}")
assert del_resp.status_code == 200
list_resp = await admin_client.get("/api/v1/contacts/")
assert list_resp.status_code == 200
ids = [c["id"] for c in list_resp.json()["items"]]
assert created_contact["id"] not in ids
class TestContactPersons:
"""Contact person management endpoints."""
@pytest.mark.asyncio
async def test_add_contact_person_returns_201(self, admin_client, created_contact):
"""POST /api/v1/contacts/:id/persons returns 201."""
response = await admin_client.post(
f"/api/v1/contacts/{created_contact['id']}/persons",
json={
"name": "Anna Schmidt",
"function": "Einkauf",
"phone": "+49 30 11122233",
"email": "anna@mueller-transport.de",
},
)
assert response.status_code == 201
data = response.json()
assert data["name"] == "Anna Schmidt"
assert data["function"] == "Einkauf"
assert data["id"] is not None
@pytest.mark.asyncio
async def test_add_contact_person_to_nonexistent_contact_returns_404(self, admin_client):
"""POST /api/v1/contacts/:nonexistent/persons returns 404."""
fake_id = uuid.uuid4()
response = await admin_client.post(
f"/api/v1/contacts/{fake_id}/persons",
json={"name": "Test Person"},
)
assert response.status_code == 404
@pytest.mark.asyncio
async def test_remove_contact_person_returns_204(self, admin_client, created_contact):
"""DELETE /api/v1/contacts/:id/persons/:person_id returns 204."""
# First add a person
add_resp = await admin_client.post(
f"/api/v1/contacts/{created_contact['id']}/persons",
json={"name": "Test Person"},
)
assert add_resp.status_code == 201
person_id = add_resp.json()["id"]
# Then remove it
del_resp = await admin_client.delete(
f"/api/v1/contacts/{created_contact['id']}/persons/{person_id}"
)
assert del_resp.status_code == 204
@pytest.mark.asyncio
async def test_remove_nonexistent_contact_person_returns_404(self, admin_client, created_contact):
"""DELETE /api/v1/contacts/:id/persons/:nonexistent returns 404."""
fake_person_id = uuid.uuid4()
response = await admin_client.delete(
f"/api/v1/contacts/{created_contact['id']}/persons/{fake_person_id}"
)
assert response.status_code == 404
@pytest.mark.asyncio
async def test_contact_detail_includes_persons(self, admin_client, created_contact):
"""GET /api/v1/contacts/:id includes contact persons in response."""
# Add a person
await admin_client.post(
f"/api/v1/contacts/{created_contact['id']}/persons",
json={"name": "Max Mustermann", "function": "Vertrieb"},
)
# Get contact detail
response = await admin_client.get(f"/api/v1/contacts/{created_contact['id']}")
assert response.status_code == 200
data = response.json()
assert len(data["contact_persons"]) >= 1
assert data["contact_persons"][0]["name"] == "Max Mustermann"
class TestContactRBAC:
"""RBAC enforcement tests."""
@pytest.mark.asyncio
async def test_list_contacts_requires_auth(self, client):
"""GET /api/v1/contacts without auth returns 401."""
response = await client.get("/api/v1/contacts/")
assert response.status_code == 401
@pytest.mark.asyncio
async def test_create_contact_as_admin_returns_201(self, admin_client, sample_contact_data):
"""POST /api/v1/contacts as admin returns 201."""
response = await admin_client.post("/api/v1/contacts/", json=sample_contact_data)
assert response.status_code == 201
@pytest.mark.asyncio
async def test_create_contact_as_verkaeufer_returns_201(self, verkaeufer_client, sample_contact_data):
"""POST /api/v1/contacts as verkaeufer returns 201."""
response = await verkaeufer_client.post("/api/v1/contacts/", json=sample_contact_data)
assert response.status_code == 201
@pytest.mark.asyncio
async def test_list_contacts_as_verkaeufer_returns_200(self, verkaeufer_client, created_contact):
"""GET /api/v1/contacts as verkaeufer returns 200 (read allowed)."""
response = await verkaeufer_client.get("/api/v1/contacts/")
assert response.status_code == 200
@pytest.mark.asyncio
async def test_update_contact_as_verkaeufer_returns_200(self, verkaeufer_client, created_contact):
"""PUT /api/v1/contacts/:id as verkaeufer returns 200."""
response = await verkaeufer_client.put(
f"/api/v1/contacts/{created_contact['id']}",
json={"company_name": "Updated by Verkaeufer"},
)
assert response.status_code == 200
@pytest.mark.asyncio
async def test_delete_contact_as_verkaeufer_returns_200(self, verkaeufer_client, sample_contact_data):
"""DELETE /api/v1/contacts/:id as verkaeufer returns 200."""
resp = await verkaeufer_client.post("/api/v1/contacts/", json=sample_contact_data)
assert resp.status_code == 201
contact_id = resp.json()["id"]
del_resp = await verkaeufer_client.delete(f"/api/v1/contacts/{contact_id}")
assert del_resp.status_code == 200
@pytest.mark.asyncio
async def test_add_person_as_verkaeufer_returns_201(self, verkaeufer_client, sample_contact_data):
"""POST /api/v1/contacts/:id/persons as verkaeufer returns 201."""
resp = await verkaeufer_client.post("/api/v1/contacts/", json=sample_contact_data)
assert resp.status_code == 201
contact_id = resp.json()["id"]
person_resp = await verkaeufer_client.post(
f"/api/v1/contacts/{contact_id}/persons",
json={"name": "Test Person"},
)
assert person_resp.status_code == 201
@pytest.mark.asyncio
async def test_remove_person_as_verkaeufer_returns_204(self, verkaeufer_client, sample_contact_data):
"""DELETE /api/v1/contacts/:id/persons/:pid as verkaeufer returns 204."""
resp = await verkaeufer_client.post("/api/v1/contacts/", json=sample_contact_data)
contact_id = resp.json()["id"]
person_resp = await verkaeufer_client.post(
f"/api/v1/contacts/{contact_id}/persons",
json={"name": "To Remove"},
)
person_id = person_resp.json()["id"]
del_resp = await verkaeufer_client.delete(
f"/api/v1/contacts/{contact_id}/persons/{person_id}"
)
assert del_resp.status_code == 204
@pytest.mark.asyncio
async def test_update_nonexistent_contact_returns_404_as_verkaeufer(self, verkaeufer_client):
"""PUT /api/v1/contacts/:nonexistent as verkaeufer returns 404."""
fake_id = uuid.uuid4()
response = await verkaeufer_client.put(
f"/api/v1/contacts/{fake_id}",
json={"company_name": "Test"},
)
assert response.status_code == 404
@pytest.mark.asyncio
async def test_delete_nonexistent_contact_returns_404_as_verkaeufer(self, verkaeufer_client):
"""DELETE /api/v1/contacts/:nonexistent as verkaeufer returns 404."""
fake_id = uuid.uuid4()
response = await verkaeufer_client.delete(f"/api/v1/contacts/{fake_id}")
assert response.status_code == 404
@pytest.mark.asyncio
async def test_add_person_to_nonexistent_contact_returns_404_as_verkaeufer(self, verkaeufer_client):
"""POST /api/v1/contacts/:nonexistent/persons as verkaeufer returns 404."""
fake_id = uuid.uuid4()
response = await verkaeufer_client.post(
f"/api/v1/contacts/{fake_id}/persons",
json={"name": "Test"},
)
assert response.status_code == 404
@pytest.mark.asyncio
async def test_remove_nonexistent_person_returns_404_as_verkaeufer(self, verkaeufer_client, sample_contact_data):
"""DELETE /api/v1/contacts/:id/persons/:nonexistent as verkaeufer returns 404."""
resp = await verkaeufer_client.post("/api/v1/contacts/", json=sample_contact_data)
contact_id = resp.json()["id"]
fake_person_id = uuid.uuid4()
response = await verkaeufer_client.delete(
f"/api/v1/contacts/{contact_id}/persons/{fake_person_id}"
)
assert response.status_code == 404
class TestContactServiceDirect:
"""Direct service-level tests for coverage."""
@pytest.mark.asyncio
async def test_service_list_contacts_with_all_filters(self, db_session):
"""Test list_contacts with all filter parameters."""
from app.services import contact_service
contact1 = Contact(
company_name="Alpha GmbH",
address_city="Berlin",
address_country="DE",
role="kaeufer",
)
contact2 = Contact(
company_name="Beta B.V.",
address_city="Amsterdam",
address_country="NL",
role="verkaeufer",
)
contact3 = Contact(
company_name="Gamma KG",
address_city="Hamburg",
address_country="DE",
role="beide",
)
db_session.add_all([contact1, contact2, contact3])
await db_session.commit()
# Test search
results, total = await contact_service.list_contacts(db_session, search="alpha")
assert total == 1
assert results[0].company_name == "Alpha GmbH"
# Test role filter (kaeufer includes beide)
results, total = await contact_service.list_contacts(db_session, role="kaeufer")
assert total == 2
# Test is_eu filter
results, total = await contact_service.list_contacts(db_session, is_eu=True)
assert total == 1
assert results[0].address_country == "NL"
# Test is_eu=false (Inland)
results, total = await contact_service.list_contacts(db_session, is_eu=False)
assert total == 2
# Test is_private filter
contact4 = Contact(
company_name="Private Person",
address_country="DE",
role="kaeufer",
is_private=True,
)
db_session.add(contact4)
await db_session.commit()
results, total = await contact_service.list_contacts(db_session, is_private=True)
assert total == 1
# Test sort descending
results, total = await contact_service.list_contacts(db_session, sort="-company_name")
names = [r.company_name for r in results]
assert names == sorted(names, reverse=True)
# Test invalid sort field falls back to created_at
results, total = await contact_service.list_contacts(db_session, sort="invalid_field")
assert total >= 4
@pytest.mark.asyncio
async def test_service_get_contact_by_id_not_found(self, db_session):
"""Test get_contact_by_id returns None for nonexistent ID."""
from app.services import contact_service
result = await contact_service.get_contact_by_id(db_session, uuid.uuid4())
assert result is None
@pytest.mark.asyncio
async def test_service_create_contact_with_persons(self, db_session):
"""Test create_contact with nested contact persons."""
from app.services import contact_service
data = {
"company_name": "Test Service GmbH",
"address_country": "DE",
"role": "kaeufer",
"contact_persons": [
{"name": "Person 1", "function": "CEO"},
{"name": "Person 2", "phone": "+49 30 123"},
],
}
contact = await contact_service.create_contact(db_session, data)
assert contact.company_name == "Test Service GmbH"
assert len(contact.contact_persons) == 2
@pytest.mark.asyncio
async def test_service_update_contact_not_found(self, db_session):
"""Test update_contact returns None for nonexistent ID."""
from app.services import contact_service
result = await contact_service.update_contact(db_session, uuid.uuid4(), {"company_name": "Test"})
assert result is None
@pytest.mark.asyncio
async def test_service_soft_delete_contact_not_found(self, db_session):
"""Test soft_delete_contact returns None for nonexistent ID."""
from app.services import contact_service
result = await contact_service.soft_delete_contact(db_session, uuid.uuid4())
assert result is None
@pytest.mark.asyncio
async def test_service_add_contact_person_not_found(self, db_session):
"""Test add_contact_person returns None for nonexistent contact."""
from app.services import contact_service
result = await contact_service.add_contact_person(db_session, uuid.uuid4(), {"name": "Test"})
assert result is None
@pytest.mark.asyncio
async def test_service_remove_contact_person_not_found(self, db_session):
"""Test remove_contact_person returns False for nonexistent person."""
from app.services import contact_service
result = await contact_service.remove_contact_person(db_session, uuid.uuid4(), uuid.uuid4())
assert result is False
@pytest.mark.asyncio
async def test_service_update_contact_success(self, db_session):
"""Test update_contact successfully updates fields."""
from app.services import contact_service
contact = Contact(
company_name="Original GmbH",
address_country="DE",
role="kaeufer",
)
db_session.add(contact)
await db_session.commit()
await db_session.refresh(contact)
updated = await contact_service.update_contact(db_session, contact.id, {"company_name": "Updated GmbH", "phone": "+49 30 999"})
assert updated.company_name == "Updated GmbH"
assert updated.phone == "+49 30 999"
@pytest.mark.asyncio
async def test_service_soft_delete_contact_success(self, db_session):
"""Test soft_delete_contact sets deleted_at."""
from app.services import contact_service
contact = Contact(
company_name="To Delete GmbH",
address_country="DE",
role="kaeufer",
)
db_session.add(contact)
await db_session.commit()
await db_session.refresh(contact)
deleted = await contact_service.soft_delete_contact(db_session, contact.id)
assert deleted.deleted_at is not None
@pytest.mark.asyncio
async def test_service_add_and_remove_contact_person(self, db_session):
"""Test add_contact_person and remove_contact_person."""
from app.services import contact_service
contact = Contact(
company_name="Person Test GmbH",
address_country="DE",
role="kaeufer",
)
db_session.add(contact)
await db_session.commit()
await db_session.refresh(contact)
person = await contact_service.add_contact_person(db_session, contact.id, {"name": "Test Person", "function": "Manager"})
assert person is not None
assert person.name == "Test Person"
removed = await contact_service.remove_contact_person(db_session, contact.id, person.id)
assert removed is True
@pytest.mark.asyncio
async def test_ust_validation_or_raise_valid(self):
"""Test validate_vat_id_or_raise with valid VAT ID."""
from app.utils.ust_validation import validate_vat_id_or_raise
result = validate_vat_id_or_raise("DE123456789")
assert result == "DE123456789"
@pytest.mark.asyncio
async def test_ust_validation_or_raise_none(self):
"""Test validate_vat_id_or_raise with None."""
from app.utils.ust_validation import validate_vat_id_or_raise
result = validate_vat_id_or_raise(None)
assert result is None
@pytest.mark.asyncio
async def test_ust_validation_or_raise_empty(self):
"""Test validate_vat_id_or_raise with empty string."""
from app.utils.ust_validation import validate_vat_id_or_raise
result = validate_vat_id_or_raise("")
assert result is None
@pytest.mark.asyncio
async def test_ust_validation_or_raise_invalid(self):
"""Test validate_vat_id_or_raise raises ValueError for invalid format."""
from app.utils.ust_validation import validate_vat_id_or_raise
with pytest.raises(ValueError, match="Invalid VAT ID format"):
validate_vat_id_or_raise("DE123")
@pytest.mark.asyncio
async def test_ust_validation_get_country_code(self):
"""Test get_country_code_from_vat_id."""
from app.utils.ust_validation import get_country_code_from_vat_id
assert get_country_code_from_vat_id("DE123456789") == "DE"
assert get_country_code_from_vat_id("at123") == "AT"
assert get_country_code_from_vat_id("") is None
assert get_country_code_from_vat_id("A") is None
assert get_country_code_from_vat_id("12") is None
@pytest.mark.asyncio
async def test_ust_validation_eu_fallback(self):
"""Test EU fallback pattern for countries without specific regex."""
from app.utils.ust_validation import validate_vat_id
# Ireland (IE) is in EU set but has no specific pattern
assert validate_vat_id("IE1234567AB") is True
# Bulgaria (BG) is in EU set but has no specific pattern
assert validate_vat_id("BG1234567890") is True
# Too short for fallback
assert validate_vat_id("BG123") is False
@@ -0,0 +1,10 @@
import { ContactDetail } from '@/components/contacts/ContactDetail';
export default async function KontaktDetailPage({
params,
}: {
params: { locale: string; id: string };
}) {
const { id } = params;
return <ContactDetail contactId={id} />;
}
@@ -0,0 +1,10 @@
import { ContactForm } from '@/components/contacts/ContactForm';
export default function NeuerKontaktPage() {
return (
<div className="space-y-6">
<h1 className="text-2xl font-bold text-text">Neuer Kontakt</h1>
<ContactForm mode="create" />
</div>
);
}
+5
View File
@@ -0,0 +1,5 @@
import { ContactList } from '@/components/contacts/ContactList';
export default function KontaktePage() {
return <ContactList />;
}
@@ -0,0 +1,246 @@
'use client';
import { useState, useEffect, useCallback } from 'react';
import { useRouter } from 'next/navigation';
import { Button } from '@/components/ui/Button';
import { Input } from '@/components/ui/Input';
import { Modal } from '@/components/ui/Modal';
import { Card } from '@/components/ui/Card';
import {
getContact,
deleteContact,
addContactPerson,
removeContactPerson,
type ContactResponse,
type ContactPersonResponse,
} from '@/lib/contacts';
interface ContactDetailProps {
contactId: string;
}
function formatDate(dateStr?: string): string {
if (!dateStr) return '—';
return new Date(dateStr).toLocaleDateString('de-DE');
}
export function ContactDetail({ contactId }: ContactDetailProps) {
const router = useRouter();
const [contact, setContact] = useState<ContactResponse | null>(null);
const [loading, setLoading] = useState(true);
const [error, setError] = useState<string | null>(null);
const [showPersonModal, setShowPersonModal] = useState(false);
const [personForm, setPersonForm] = useState({ name: '', function: '', phone: '', email: '' });
const [personError, setPersonError] = useState<string | null>(null);
const [addingPerson, setAddingPerson] = useState(false);
const fetchContact = useCallback(async () => {
setLoading(true);
setError(null);
try {
const data = await getContact(contactId);
setContact(data);
} catch (err: unknown) {
const apiErr = err as { error?: { message?: string } };
setError(apiErr?.error?.message || 'Failed to load contact');
} finally {
setLoading(false);
}
}, [contactId]);
useEffect(() => {
fetchContact();
}, [fetchContact]);
const handleDelete = async () => {
if (!contact) return;
if (!confirm('Diesen Kontakt löschen?')) return;
try {
await deleteContact(contact.id);
router.push('/de/kontakte');
} catch (err: unknown) {
const apiErr = err as { error?: { message?: string } };
setError(apiErr?.error?.message || 'Failed to delete contact');
}
};
const handleAddPerson = async (e: React.FormEvent) => {
e.preventDefault();
if (!contact || !personForm.name.trim()) return;
setAddingPerson(true);
setPersonError(null);
try {
await addContactPerson(contact.id, {
name: personForm.name,
function: personForm.function || undefined,
phone: personForm.phone || undefined,
email: personForm.email || undefined,
});
setShowPersonModal(false);
setPersonForm({ name: '', function: '', phone: '', email: '' });
await fetchContact();
} catch (err: unknown) {
const apiErr = err as { error?: { message?: string } };
setPersonError(apiErr?.error?.message || 'Failed to add person');
} finally {
setAddingPerson(false);
}
};
const handleRemovePerson = async (personId: string) => {
if (!contact) return;
if (!confirm('Diese Kontaktperson entfernen?')) return;
try {
await removeContactPerson(contact.id, personId);
await fetchContact();
} catch (err: unknown) {
const apiErr = err as { error?: { message?: string } };
setError(apiErr?.error?.message || 'Failed to remove person');
}
};
if (loading) {
return (
<div data-testid="contact-detail-loading" className="text-center py-8 text-text-muted">
Loading contact...
</div>
);
}
if (error) {
return (
<div data-testid="contact-detail-error" className="p-4 bg-error/10 text-error rounded-lg">
{error}
</div>
);
}
if (!contact) {
return (
<div data-testid="contact-detail-not-found" className="text-center py-8 text-text-muted">
Contact not found
</div>
);
}
const fields: { label: string; value: string | number | undefined | null }[] = [
{ label: 'Firmenname', value: contact.company_name },
{ label: 'Rechtsform', value: contact.legal_form },
{ label: 'Straße', value: contact.address_street },
{ label: 'PLZ', value: contact.address_zip },
{ label: 'Stadt', value: contact.address_city },
{ label: 'Land', value: contact.address_country },
{ label: 'USt-IdNr.', value: contact.vat_id },
{ label: 'USt-Status', value: contact.vat_id_status },
{ label: 'Telefon', value: contact.phone },
{ label: 'E-Mail', value: contact.email },
{ label: 'Website', value: contact.website },
{ label: 'Rolle', value: contact.role },
{ label: 'Privat', value: contact.is_private ? 'Ja' : 'Nein' },
{ label: 'Erstellt am', value: formatDate(contact.created_at) },
{ label: 'Aktualisiert am', value: formatDate(contact.updated_at) },
];
return (
<div data-testid="contact-detail" className="space-y-6">
<div className="flex items-center justify-between">
<h1 data-testid="contact-detail-title" className="text-2xl font-bold text-text">
{contact.company_name}
</h1>
<div className="flex gap-2">
<Button variant="secondary" onClick={() => router.push(`/de/kontakte/${contact.id}/bearbeiten`)}>
Bearbeiten
</Button>
<Button variant="danger" onClick={handleDelete} data-testid="delete-button">
Löschen
</Button>
</div>
</div>
<div data-testid="contact-detail-fields" className="grid grid-cols-2 md:grid-cols-3 gap-4 p-6 bg-surface rounded-lg border border-border">
{fields.map(field => (
<div key={field.label} className="space-y-1">
<dt className="text-sm font-medium text-text-muted">{field.label}</dt>
<dd data-testid={`field-${field.label.toLowerCase().replace(/\s+/g, '_')}`} className="text-text">
{field.value !== null && field.value !== undefined && field.value !== '' ? field.value : '—'}
</dd>
</div>
))}
</div>
<Card title="Kontaktpersonen">
<div data-testid="contact-persons-section" className="space-y-3">
{contact.contact_persons && contact.contact_persons.length > 0 ? (
contact.contact_persons.map((person: ContactPersonResponse) => (
<div key={person.id} data-testid={`contact-person-${person.id}`} className="flex items-center justify-between p-3 bg-background/50 rounded-lg">
<div className="space-y-1">
<p className="font-medium text-text">{person.name}</p>
{person.function && <p className="text-sm text-text-muted">{person.function}</p>}
{person.phone && <p className="text-sm text-text-muted">Tel: {person.phone}</p>}
{person.email && <p className="text-sm text-text-muted">E-Mail: {person.email}</p>}
</div>
<Button
variant="danger"
onClick={() => handleRemovePerson(person.id)}
data-testid={`remove-person-${person.id}`}
>
Entfernen
</Button>
</div>
))
) : (
<p data-testid="no-contact-persons" className="text-text-muted text-center py-4">
Keine Kontaktpersonen vorhanden.
</p>
)}
<Button onClick={() => setShowPersonModal(true)} data-testid="add-person-btn">
+ Kontaktperson hinzufügen
</Button>
</div>
</Card>
<Modal open={showPersonModal} onClose={() => setShowPersonModal(false)} title="Kontaktperson hinzufügen">
<form data-testid="person-form" onSubmit={handleAddPerson} className="space-y-4">
{personError && (
<div data-testid="person-form-error" className="p-3 bg-error/10 text-error rounded-lg">
{personError}
</div>
)}
<Input
label="Name *"
data-testid="person-input-name"
value={personForm.name}
onChange={e => setPersonForm(prev => ({ ...prev, name: e.target.value }))}
/>
<Input
label="Funktion"
data-testid="person-input-function"
value={personForm.function}
onChange={e => setPersonForm(prev => ({ ...prev, function: e.target.value }))}
/>
<Input
label="Telefon"
data-testid="person-input-phone"
value={personForm.phone}
onChange={e => setPersonForm(prev => ({ ...prev, phone: e.target.value }))}
/>
<Input
label="E-Mail"
data-testid="person-input-email"
type="email"
value={personForm.email}
onChange={e => setPersonForm(prev => ({ ...prev, email: e.target.value }))}
/>
<div className="flex gap-3">
<Button type="submit" loading={addingPerson} data-testid="person-submit">
Hinzufügen
</Button>
<Button type="button" variant="ghost" onClick={() => setShowPersonModal(false)}>
Abbrechen
</Button>
</div>
</form>
</Modal>
</div>
);
}
@@ -0,0 +1,290 @@
'use client';
import { useState } from 'react';
import { useRouter } from 'next/navigation';
import { Button } from '@/components/ui/Button';
import { Input } from '@/components/ui/Input';
import {
createContact,
updateContact,
validateVatIdFormat,
type ContactResponse,
type ContactCreateData,
} from '@/lib/contacts';
const ROLE_OPTIONS = ['kaeufer', 'verkaeufer', 'beide'];
const LEGAL_FORMS = ['GmbH', 'AG', 'KG', 'OHG', 'GbR', 'e.K.', 'UG', 'SE', 'Einzelunternehmen'];
const COUNTRY_OPTIONS = [
'DE', 'AT', 'BE', 'BG', 'CY', 'CZ', 'DK', 'EE', 'ES', 'FI', 'FR', 'GR',
'HR', 'HU', 'IE', 'IT', 'LT', 'LU', 'LV', 'MT', 'NL', 'PL', 'PT', 'RO',
'SE', 'SI', 'SK',
];
interface ContactFormProps {
contact?: ContactResponse;
mode?: 'create' | 'edit';
}
interface FormErrors {
[key: string]: string | undefined;
}
export function ContactForm({ contact, mode = 'create' }: ContactFormProps) {
const router = useRouter();
const [loading, setLoading] = useState(false);
const [errors, setErrors] = useState<FormErrors>({});
const [submitError, setSubmitError] = useState<string | null>(null);
const [formData, setFormData] = useState<ContactCreateData>({
company_name: contact?.company_name || '',
legal_form: contact?.legal_form,
address_street: contact?.address_street,
address_zip: contact?.address_zip,
address_city: contact?.address_city,
address_country: contact?.address_country || 'DE',
vat_id: contact?.vat_id,
phone: contact?.phone,
email: contact?.email,
website: contact?.website,
role: contact?.role || 'kaeufer',
is_private: contact?.is_private || false,
});
const isEu = formData.address_country !== 'DE';
const validate = (): boolean => {
const newErrors: FormErrors = {};
if (!formData.company_name || formData.company_name.trim().length === 0) {
newErrors.company_name = 'Company name is required';
}
if (!formData.role) {
newErrors.role = 'Role is required';
}
if (!formData.address_country || formData.address_country.length !== 2) {
newErrors.address_country = 'Country code is required (2 letters)';
}
// VAT ID validation
if (formData.vat_id) {
const vatError = validateVatIdFormat(formData.vat_id, formData.address_country);
if (vatError) {
newErrors.vat_id = vatError;
}
}
// EU contacts should have a VAT ID (warning, not error)
if (isEu && !formData.vat_id) {
// Soft warning - don't block submission
}
setErrors(newErrors);
return Object.keys(newErrors).length === 0;
};
const handleChange = (field: keyof ContactCreateData, value: string | boolean | undefined) => {
setFormData(prev => ({ ...prev, [field]: value }));
if (errors[field]) {
setErrors(prev => { const next = { ...prev }; delete next[field]; return next; });
}
};
const handleSubmit = async (e: React.FormEvent) => {
e.preventDefault();
if (!validate()) return;
setLoading(true);
setSubmitError(null);
try {
const submitData = { ...formData };
// Remove undefined empty strings
Object.keys(submitData).forEach(key => {
if (submitData[key as keyof ContactCreateData] === '') {
(submitData as Record<string, unknown>)[key] = undefined;
}
});
if (mode === 'edit' && contact) {
await updateContact(contact.id, submitData);
router.push(`/de/kontakte/${contact.id}`);
} else {
const created = await createContact(submitData);
router.push(`/de/kontakte/${created.id}`);
}
} catch (err: unknown) {
const apiErr = err as { error?: { message?: string } };
setSubmitError(apiErr?.error?.message || 'Failed to save contact');
} finally {
setLoading(false);
}
};
const inputClass = 'w-full px-3 py-2 border rounded-lg bg-surface text-text border-border focus:outline-none focus:ring-2 focus:ring-primary';
return (
<form data-testid="contact-form" onSubmit={handleSubmit} className="space-y-6">
{submitError && (
<div data-testid="form-submit-error" className="p-4 bg-error/10 text-error rounded-lg">
{submitError}
</div>
)}
{/* EU / Inland Toggle */}
<div data-testid="eu-inland-toggle" className="flex gap-3 p-4 bg-surface rounded-lg border border-border">
<label className="flex items-center gap-2 cursor-pointer">
<input
type="radio"
data-testid="toggle-inland"
name="region"
value="DE"
checked={formData.address_country === 'DE'}
onChange={() => handleChange('address_country', 'DE')}
/>
<span className="text-text font-medium">Inland (DE)</span>
</label>
<label className="flex items-center gap-2 cursor-pointer">
<input
type="radio"
data-testid="toggle-eu"
name="region"
value="EU"
checked={formData.address_country !== 'DE'}
onChange={() => handleChange('address_country', 'AT')}
/>
<span className="text-text font-medium">EU (Ausland)</span>
</label>
</div>
<div className="grid grid-cols-2 gap-4">
<Input
label="Firmenname *"
data-testid="input-company_name"
value={formData.company_name}
onChange={e => handleChange('company_name', e.target.value)}
error={errors.company_name}
/>
<div>
<label className="block text-sm font-medium text-text mb-1">Rechtsform</label>
<select
data-testid="input-legal_form"
className={inputClass}
value={formData.legal_form || ''}
onChange={e => handleChange('legal_form', e.target.value || undefined)}
>
<option value=""></option>
{LEGAL_FORMS.map(lf => <option key={lf} value={lf}>{lf}</option>)}
</select>
</div>
</div>
<div className="grid grid-cols-2 gap-4">
<div>
<label className="block text-sm font-medium text-text mb-1">Rolle *</label>
<select
data-testid="input-role"
className={inputClass}
value={formData.role}
onChange={e => handleChange('role', e.target.value)}
>
{ROLE_OPTIONS.map(r => <option key={r} value={r}>{r}</option>)}
</select>
{errors.role && <p className="mt-1 text-sm text-error">{errors.role}</p>}
</div>
<div>
<label className="block text-sm font-medium text-text mb-1">Land *</label>
<select
data-testid="input-address_country"
className={inputClass}
value={formData.address_country}
onChange={e => handleChange('address_country', e.target.value)}
>
{COUNTRY_OPTIONS.map(c => <option key={c} value={c}>{c}</option>)}
</select>
{errors.address_country && <p className="mt-1 text-sm text-error">{errors.address_country}</p>}
</div>
</div>
<div className="grid grid-cols-3 gap-4">
<Input
label="Straße"
data-testid="input-address_street"
value={formData.address_street || ''}
onChange={e => handleChange('address_street', e.target.value || undefined)}
/>
<Input
label="PLZ"
data-testid="input-address_zip"
value={formData.address_zip || ''}
onChange={e => handleChange('address_zip', e.target.value || undefined)}
/>
<Input
label="Stadt"
data-testid="input-address_city"
value={formData.address_city || ''}
onChange={e => handleChange('address_city', e.target.value || undefined)}
/>
</div>
<div className="grid grid-cols-2 gap-4">
<div>
<Input
label={isEu ? 'USt-IdNr. (empfohlen)' : 'USt-IdNr.'}
data-testid="input-vat_id"
value={formData.vat_id || ''}
onChange={e => handleChange('vat_id', e.target.value || undefined)}
error={errors.vat_id}
placeholder={isEu ? 'z.B. ATU12345678' : 'z.B. DE123456789'}
/>
{isEu && !formData.vat_id && (
<p data-testid="vat-id-hint" className="mt-1 text-sm text-text-muted">
Für EU-Kontakte wird eine USt-IdNr. empfohlen.
</p>
)}
</div>
<Input
label="Telefon"
data-testid="input-phone"
value={formData.phone || ''}
onChange={e => handleChange('phone', e.target.value || undefined)}
/>
</div>
<div className="grid grid-cols-2 gap-4">
<Input
label="E-Mail"
data-testid="input-email"
type="email"
value={formData.email || ''}
onChange={e => handleChange('email', e.target.value || undefined)}
/>
<Input
label="Website"
data-testid="input-website"
value={formData.website || ''}
onChange={e => handleChange('website', e.target.value || undefined)}
/>
</div>
<div className="flex items-center gap-2">
<label className="flex items-center gap-2 cursor-pointer">
<input
type="checkbox"
data-testid="input-is_private"
checked={formData.is_private || false}
onChange={e => handleChange('is_private', e.target.checked)}
/>
<span className="text-sm text-text">Privatkontakt</span>
</label>
</div>
<div className="flex gap-3">
<Button type="submit" loading={loading} data-testid="submit-button">
{mode === 'edit' ? 'Kontakt aktualisieren' : 'Kontakt erstellen'}
</Button>
<Button type="button" variant="ghost" onClick={() => router.back()}>
Abbrechen
</Button>
</div>
</form>
);
}
@@ -0,0 +1,228 @@
'use client';
import { useState, useEffect, useCallback } from 'react';
import { useRouter } from 'next/navigation';
import { Table } from '@/components/ui/Table';
import { Button } from '@/components/ui/Button';
import { Input } from '@/components/ui/Input';
import {
listContacts,
type ContactResponse,
type ContactListParams,
} from '@/lib/contacts';
import type { PaginatedResponse } from '@/lib/api';
const ROLE_OPTIONS = ['kaeufer', 'verkaeufer', 'beide'];
const SORT_OPTIONS = [
{ value: '-created_at', label: 'Newest First' },
{ value: 'created_at', label: 'Oldest First' },
{ value: 'company_name', label: 'Company A-Z' },
{ value: '-company_name', label: 'Company Z-A' },
{ value: 'address_city', label: 'City A-Z' },
];
export function ContactList() {
const router = useRouter();
const [contacts, setContacts] = useState<ContactResponse[]>([]);
const [total, setTotal] = useState(0);
const [page, setPage] = useState(1);
const [pageSize] = useState(20);
const [loading, setLoading] = useState(false);
const [error, setError] = useState<string | null>(null);
const [filters, setFilters] = useState<ContactListParams>({
page: 1,
page_size: 20,
});
const fetchContacts = useCallback(async () => {
setLoading(true);
setError(null);
try {
const data: PaginatedResponse<ContactResponse> = await listContacts(filters);
setContacts(data.items);
setTotal(data.total);
setPage(data.page);
} catch (err: unknown) {
const apiErr = err as { error?: { message?: string } };
setError(apiErr?.error?.message || 'Failed to load contacts');
} finally {
setLoading(false);
}
}, [filters]);
useEffect(() => {
fetchContacts();
}, [fetchContacts]);
const handleFilterChange = (key: keyof ContactListParams, value: string) => {
setFilters(prev => ({
...prev,
page: 1,
[key]: value || undefined,
}));
};
const handleEuToggle = (value: string) => {
if (value === '') {
setFilters(prev => ({ ...prev, page: 1, is_eu: undefined }));
} else {
setFilters(prev => ({ ...prev, page: 1, is_eu: value === 'eu' }));
}
};
const handlePageChange = (newPage: number) => {
setFilters(prev => ({ ...prev, page: newPage }));
};
const columns = [
{
key: 'company_name',
label: 'Company',
render: (row: ContactResponse) => (
<button
data-testid={`contact-row-${row.id}`}
onClick={() => router.push(`/de/kontakte/${row.id}`)}
className="text-primary hover:underline"
>
{row.company_name}
</button>
),
},
{ key: 'address_city', label: 'City' },
{ key: 'address_country', label: 'Country' },
{ key: 'role', label: 'Role' },
{
key: 'vat_id',
label: 'VAT ID',
render: (row: ContactResponse) => row.vat_id || '—',
},
{
key: 'vat_id_status',
label: 'VAT Status',
render: (row: ContactResponse) => (
<span
className={`px-2 py-1 rounded text-xs ${
row.vat_id_status === 'geprueft' || row.vat_id_status === 'manuell_bestaetigt'
? 'bg-success/20 text-success'
: row.vat_id_status === 'ungueltig'
? 'bg-error/20 text-error'
: 'bg-secondary/20 text-secondary'
}`}
>
{row.vat_id_status}
</span>
),
},
{
key: 'contact_persons',
label: 'Persons',
render: (row: ContactResponse) => String(row.contact_persons?.length || 0),
},
];
const totalPages = Math.ceil(total / pageSize);
return (
<div data-testid="contact-list" className="space-y-4">
<div className="flex items-center justify-between">
<h1 className="text-2xl font-bold text-text">Kontakte</h1>
<Button onClick={() => router.push('/de/kontakte/neu')} data-testid="new-contact-btn">
+ Neuer Kontakt
</Button>
</div>
<div data-testid="contact-filters" className="flex flex-wrap gap-3 p-4 bg-surface rounded-lg border border-border">
<div className="w-48">
<label className="block text-sm font-medium text-text mb-1">Search</label>
<Input
data-testid="filter-search"
type="text"
placeholder="Company, city, email..."
value={filters.search || ''}
onChange={e => handleFilterChange('search', e.target.value)}
/>
</div>
<div className="w-40">
<label className="block text-sm font-medium text-text mb-1">Role</label>
<select
data-testid="filter-role"
className="w-full px-3 py-2 border rounded-lg bg-surface text-text border-border"
value={filters.role || ''}
onChange={e => handleFilterChange('role', e.target.value)}
>
<option value="">All</option>
{ROLE_OPTIONS.map(r => (
<option key={r} value={r}>{r}</option>
))}
</select>
</div>
<div className="w-40">
<label className="block text-sm font-medium text-text mb-1">Region</label>
<select
data-testid="filter-eu"
className="w-full px-3 py-2 border rounded-lg bg-surface text-text border-border"
value={filters.is_eu === undefined ? '' : filters.is_eu ? 'eu' : 'inland'}
onChange={e => handleEuToggle(e.target.value)}
>
<option value="">All</option>
<option value="inland">Inland (DE)</option>
<option value="eu">EU (non-DE)</option>
</select>
</div>
<div className="w-40">
<label className="block text-sm font-medium text-text mb-1">Sort</label>
<select
data-testid="filter-sort"
className="w-full px-3 py-2 border rounded-lg bg-surface text-text border-border"
value={filters.sort || ''}
onChange={e => handleFilterChange('sort', e.target.value)}
>
<option value="">Default</option>
{SORT_OPTIONS.map(s => (
<option key={s.value} value={s.value}>{s.label}</option>
))}
</select>
</div>
</div>
{error && (
<div data-testid="contact-error" className="p-4 bg-error/10 text-error rounded-lg">
{error}
</div>
)}
{loading ? (
<div data-testid="contact-loading" className="text-center py-8 text-text-muted">
Loading contacts...
</div>
) : (
<Table columns={columns} data={contacts} rowKey={row => row.id} />
)}
{totalPages > 1 && (
<div data-testid="contact-pagination" className="flex items-center justify-between">
<span className="text-sm text-text-muted">
Page {page} of {totalPages} ({total} total)
</span>
<div className="flex gap-2">
<Button
variant="secondary"
disabled={page <= 1}
onClick={() => handlePageChange(page - 1)}
>
Previous
</Button>
<Button
variant="secondary"
disabled={page >= totalPages}
onClick={() => handlePageChange(page + 1)}
>
Next
</Button>
</div>
</div>
)}
</div>
);
}
+163
View File
@@ -0,0 +1,163 @@
import { apiFetch, type PaginatedResponse } from './api';
export interface ContactPersonResponse {
id: string;
contact_id: string;
name: string;
function?: string;
phone?: string;
email?: string;
created_at?: string;
}
export interface ContactResponse {
id: string;
company_name: string;
legal_form?: string;
address_street?: string;
address_zip?: string;
address_city?: string;
address_country: string;
vat_id?: string;
phone?: string;
email?: string;
website?: string;
role: string;
vat_id_status: string;
is_private: boolean;
created_at?: string;
updated_at?: string;
deleted_at?: string | null;
contact_persons: ContactPersonResponse[];
}
export interface ContactCreateData {
company_name: string;
legal_form?: string;
address_street?: string;
address_zip?: string;
address_city?: string;
address_country: string;
vat_id?: string;
phone?: string;
email?: string;
website?: string;
role: string;
is_private?: boolean;
contact_persons?: ContactPersonCreateData[];
}
export interface ContactUpdateData {
company_name?: string;
legal_form?: string;
address_street?: string;
address_zip?: string;
address_city?: string;
address_country?: string;
vat_id?: string;
phone?: string;
email?: string;
website?: string;
role?: string;
is_private?: boolean;
}
export interface ContactPersonCreateData {
name: string;
function?: string;
phone?: string;
email?: string;
}
export interface ContactListParams {
page?: number;
page_size?: number;
search?: string;
role?: string;
is_eu?: boolean;
is_private?: boolean;
sort?: string;
}
export async function listContacts(params: ContactListParams = {}): Promise<PaginatedResponse<ContactResponse>> {
const query = new URLSearchParams();
if (params.page) query.set('page', String(params.page));
if (params.page_size) query.set('page_size', String(params.page_size));
if (params.search) query.set('search', params.search);
if (params.role) query.set('role', params.role);
if (params.is_eu !== undefined) query.set('is_eu', String(params.is_eu));
if (params.is_private !== undefined) query.set('is_private', String(params.is_private));
if (params.sort) query.set('sort', params.sort);
const qs = query.toString();
return apiFetch<PaginatedResponse<ContactResponse>>(`/contacts/${qs ? `?${qs}` : ''}`);
}
export async function getContact(id: string): Promise<ContactResponse> {
return apiFetch<ContactResponse>(`/contacts/${id}`);
}
export async function createContact(data: ContactCreateData): Promise<ContactResponse> {
return apiFetch<ContactResponse>('/contacts/', {
method: 'POST',
body: JSON.stringify(data),
});
}
export async function updateContact(id: string, data: ContactUpdateData): Promise<ContactResponse> {
return apiFetch<ContactResponse>(`/contacts/${id}`, {
method: 'PUT',
body: JSON.stringify(data),
});
}
export async function deleteContact(id: string): Promise<ContactResponse> {
return apiFetch<ContactResponse>(`/contacts/${id}`, { method: 'DELETE' });
}
export async function addContactPerson(contactId: string, data: ContactPersonCreateData): Promise<ContactPersonResponse> {
return apiFetch<ContactPersonResponse>(`/contacts/${contactId}/persons`, {
method: 'POST',
body: JSON.stringify(data),
});
}
export async function removeContactPerson(contactId: string, personId: string): Promise<void> {
await apiFetch<void>(`/contacts/${contactId}/persons/${personId}`, { method: 'DELETE' });
}
/**
* Validate VAT ID format on the frontend (mirrors backend ust_validation.py).
* Returns error message or null if valid.
*/
export function validateVatIdFormat(vatId: string, countryCode: string): string | null {
if (!vatId) return null;
const normalized = vatId.trim().toUpperCase().replace(/\s+/g, '');
const patterns: Record<string, RegExp> = {
DE: /^DE\d{9}$/,
AT: /^ATU\d{8}$/,
FR: /^FR[A-Za-z0-9]{2}\d{9}$/,
NL: /^NL\d{9}B\d{2}$/,
PL: /^PL\d{10}$/,
CZ: /^CZ\d{8,10}$/,
IT: /^IT\d{11}$/,
ES: /^ES[A-Za-z0-9]\d{7}[A-Za-z0-9]$/,
BE: /^BE\d{10}$/,
DK: /^DK\d{8}$/,
SE: /^SE\d{10}$/,
};
const euCountries = new Set([
'AT', 'BE', 'BG', 'CY', 'CZ', 'DE', 'DK', 'EE', 'ES', 'FI', 'FR', 'GR',
'HR', 'HU', 'IE', 'IT', 'LT', 'LU', 'LV', 'MT', 'NL', 'PL', 'PT', 'RO',
'SE', 'SI', 'SK',
]);
const cc = normalized.substring(0, 2);
if (!cc.match(/^[A-Z]{2}$/)) return 'Invalid country code';
const pattern = patterns[cc];
if (pattern) {
return pattern.test(normalized) ? null : `Invalid VAT ID format for ${cc}`;
}
if (euCountries.has(cc)) {
return /^[A-Z]{2}[A-Za-z0-9]{5,15}$/.test(normalized) ? null : `Invalid VAT ID format for ${cc}`;
}
return `VAT ID validation not supported for country ${cc}`;
}
+243
View File
@@ -0,0 +1,243 @@
import { describe, it, expect, vi, beforeEach } from 'vitest';
import { render, screen, fireEvent, waitFor } from '@testing-library/react';
import { ContactList } from '@/components/contacts/ContactList';
import { ContactForm } from '@/components/contacts/ContactForm';
import { validateVatIdFormat } from '@/lib/contacts';
// Mock the contacts API module
vi.mock('@/lib/contacts', async (importOriginal) => {
const actual = await importOriginal<typeof import('@/lib/contacts')>();
return {
...actual,
listContacts: vi.fn(),
createContact: vi.fn(),
updateContact: vi.fn(),
};
});
import { listContacts, createContact } from '@/lib/contacts';
const mockContacts = {
items: [
{
id: 'test-id-1',
company_name: 'Müller Transport GmbH',
address_city: 'Berlin',
address_country: 'DE',
role: 'kaeufer',
vat_id: 'DE123456789',
vat_id_status: 'ungeprueft',
is_private: false,
contact_persons: [],
},
{
id: 'test-id-2',
company_name: 'Van der Berg B.V.',
address_city: 'Amsterdam',
address_country: 'NL',
role: 'verkaeufer',
vat_id: 'NL123456789B01',
vat_id_status: 'geprueft',
is_private: false,
contact_persons: [{ id: 'p1', contact_id: 'test-id-2', name: 'Jan' }],
},
],
total: 2,
page: 1,
page_size: 20,
};
describe('ContactList', () => {
beforeEach(() => {
vi.clearAllMocks();
});
it('renders contact list with search and filter controls', async () => {
vi.mocked(listContacts).mockResolvedValue(mockContacts);
render(<ContactList />);
await waitFor(() => {
expect(screen.getByTestId('contact-list')).toBeInTheDocument();
});
expect(screen.getByTestId('filter-search')).toBeInTheDocument();
expect(screen.getByTestId('filter-role')).toBeInTheDocument();
expect(screen.getByTestId('filter-eu')).toBeInTheDocument();
expect(screen.getByTestId('filter-sort')).toBeInTheDocument();
});
it('displays contacts in table after loading', async () => {
vi.mocked(listContacts).mockResolvedValue(mockContacts);
render(<ContactList />);
await waitFor(() => {
expect(screen.getByText('Müller Transport GmbH')).toBeInTheDocument();
});
expect(screen.getByText('Van der Berg B.V.')).toBeInTheDocument();
expect(screen.getByText('Berlin')).toBeInTheDocument();
expect(screen.getByText('Amsterdam')).toBeInTheDocument();
});
it('shows new contact button', async () => {
vi.mocked(listContacts).mockResolvedValue(mockContacts);
render(<ContactList />);
await waitFor(() => {
expect(screen.getByTestId('new-contact-btn')).toBeInTheDocument();
});
});
it('shows error message on API failure', async () => {
vi.mocked(listContacts).mockRejectedValue({
error: { message: 'Network error' },
});
render(<ContactList />);
await waitFor(() => {
expect(screen.getByTestId('contact-error')).toBeInTheDocument();
});
expect(screen.getByText('Network error')).toBeInTheDocument();
});
it('shows loading state initially', async () => {
vi.mocked(listContacts).mockImplementation(
() => new Promise((resolve) => setTimeout(() => resolve(mockContacts), 100))
);
render(<ContactList />);
expect(screen.getByTestId('contact-loading')).toBeInTheDocument();
await waitFor(() => {
expect(screen.getByText('Müller Transport GmbH')).toBeInTheDocument();
});
});
});
describe('ContactForm', () => {
beforeEach(() => {
vi.clearAllMocks();
});
it('renders form with all required fields', () => {
render(<ContactForm mode="create" />);
expect(screen.getByTestId('contact-form')).toBeInTheDocument();
expect(screen.getByTestId('input-company_name')).toBeInTheDocument();
expect(screen.getByTestId('input-role')).toBeInTheDocument();
expect(screen.getByTestId('input-address_country')).toBeInTheDocument();
expect(screen.getByTestId('input-vat_id')).toBeInTheDocument();
expect(screen.getByTestId('input-phone')).toBeInTheDocument();
expect(screen.getByTestId('input-email')).toBeInTheDocument();
expect(screen.getByTestId('submit-button')).toBeInTheDocument();
});
it('shows EU/Inland toggle', () => {
render(<ContactForm mode="create" />);
expect(screen.getByTestId('eu-inland-toggle')).toBeInTheDocument();
expect(screen.getByTestId('toggle-inland')).toBeInTheDocument();
expect(screen.getByTestId('toggle-eu')).toBeInTheDocument();
});
it('defaults to Inland (DE) and shows DE placeholder for VAT ID', () => {
render(<ContactForm mode="create" />);
const inlandRadio = screen.getByTestId('toggle-inland') as HTMLInputElement;
expect(inlandRadio.checked).toBe(true);
});
it('switches to EU mode and shows VAT ID hint when EU is selected', () => {
render(<ContactForm mode="create" />);
const euRadio = screen.getByTestId('toggle-eu');
fireEvent.click(euRadio);
expect(screen.getByTestId('vat-id-hint')).toBeInTheDocument();
});
it('validates VAT ID format and shows error for invalid DE VAT', () => {
render(<ContactForm mode="create" />);
const vatInput = screen.getByTestId('input-vat_id') as HTMLInputElement;
fireEvent.change(vatInput, { target: { value: 'DE123' } });
const submitBtn = screen.getByTestId('submit-button');
fireEvent.click(submitBtn);
expect(screen.getByText(/Invalid VAT ID format/)).toBeInTheDocument();
});
it('accepts valid DE VAT ID without error', async () => {
vi.mocked(createContact).mockResolvedValue({
id: 'new-id',
company_name: 'Test GmbH',
address_country: 'DE',
role: 'kaeufer',
vat_id_status: 'ungeprueft',
is_private: false,
contact_persons: [],
});
render(<ContactForm mode="create" />);
const companyInput = screen.getByTestId('input-company_name');
fireEvent.change(companyInput, { target: { value: 'Test GmbH' } });
const vatInput = screen.getByTestId('input-vat_id');
fireEvent.change(vatInput, { target: { value: 'DE123456789' } });
const submitBtn = screen.getByTestId('submit-button');
fireEvent.click(submitBtn);
await waitFor(() => {
expect(createContact).toHaveBeenCalled();
});
});
it('requires company name', () => {
render(<ContactForm mode="create" />);
const submitBtn = screen.getByTestId('submit-button');
fireEvent.click(submitBtn);
expect(screen.getByText('Company name is required')).toBeInTheDocument();
});
});
describe('validateVatIdFormat', () => {
it('returns null for empty VAT ID', () => {
expect(validateVatIdFormat('', 'DE')).toBeNull();
});
it('returns null for valid DE VAT ID', () => {
expect(validateVatIdFormat('DE123456789', 'DE')).toBeNull();
});
it('returns error for invalid DE VAT ID (too short)', () => {
const result = validateVatIdFormat('DE12345678', 'DE');
expect(result).toContain('Invalid');
});
it('returns null for valid AT VAT ID', () => {
expect(validateVatIdFormat('ATU12345678', 'AT')).toBeNull();
});
it('returns null for valid NL VAT ID', () => {
expect(validateVatIdFormat('NL123456789B01', 'NL')).toBeNull();
});
it('returns error for non-EU country', () => {
const result = validateVatIdFormat('US123456789', 'US');
expect(result).toContain('not supported');
});
it('handles lowercase input', () => {
expect(validateVatIdFormat('de123456789', 'DE')).toBeNull();
});
it('handles spaces in VAT ID', () => {
expect(validateVatIdFormat('DE 123 456 789', 'DE')).toBeNull();
});
});
+44 -77
View File
@@ -1,104 +1,71 @@
# Test Report T02: Vehicle Management + mobile.de Push + Vehicle UI # Test Report T04: Kontakt-/Kundenverwaltung + Contact UI
**Date**: 2026-07-14
**Task**: T02
**Status**: PASSED
## Backend Tests ## Backend Tests
**Command**: `cd backend && python -m pytest tests/test_vehicles.py tests/test_mobilede.py tests/test_vehicles_extra.py --cov=app.services.vehicle_service --cov=app.services.mobilede_service --cov=app.routers.vehicles --cov-report=term-missing -v` **Command:** `cd backend && python -m pytest tests/test_contacts.py --cov=app.routers.contacts --cov=app.services.contact_service --cov=app.utils.ust_validation --cov=app.models.contact --cov=app.schemas.contact --cov-report=term-missing -v`
**Result**: 73 passed in 18.84s **Result:** 73 passed, 0 failed
### Coverage Report ### Coverage
| Module | Stmts | Miss | Cover | | Module | Stmts | Miss | Cover |
|--------|-------|------|-------| |--------|-------|------|-------|
| app/routers/vehicles.py | 63 | 25 | 60% | | app/models/contact.py | 55 | 4 | 93% |
| app/services/mobilede_service.py | 122 | 24 | 80% | | app/routers/contacts.py | 51 | 17 | 67% |
| app/services/vehicle_service.py | 86 | 1 | 99% | | app/schemas/contact.py | 98 | 6 | 94% |
| **TOTAL** | **271** | **50** | **82%** | | app/services/contact_service.py | 100 | 1 | 99% |
| app/utils/ust_validation.py | 31 | 2 | 94% |
| **TOTAL** | **335** | **30** | **91%** |
**Coverage target**: >= 80% → **MET** (82% total) **Coverage target:** >= 80% → **PASSED (91%)**
### Test Categories ### Test Categories
#### test_vehicles.py (27 tests) - **TestUstValidation** (13 tests): DE/EU VAT ID format validation, edge cases
- TestVehicleList: 7 tests (pagination, filter by type/availability/price, sort, search, auth required) - **TestContactList** (10 tests): Pagination, role filter (beide included), is_eu filter, search, sort
- TestVehicleCreate: 6 tests (201 on valid, 422 on missing make/fin/short fin, 409 on duplicate FIN, auto-compute power_hp) - **TestContactDetail** (3 tests): Detail with contact persons, 404 for nonexistent, 404 after soft-delete
- TestVehicleDetail: 2 tests (200 on found, 404 on nonexistent) - **TestContactCreate** (7 tests): Valid create, invalid VAT ID 422, missing fields 422, nested contact persons
- TestVehicleUpdate: 3 tests (200 on update, 404 on nonexistent, 400 on no fields) - **TestContactUpdate** (4 tests): Valid update, 404 nonexistent, invalid VAT 422, no fields 400
- TestVehicleDelete: 4 tests (200 with deleted_at, 404 on nonexistent, not in list after delete, 404 on detail after delete) - **TestContactDelete** (3 tests): Soft-delete, 404 nonexistent, not in list after delete
- TestMobileDePush: 2 tests (202 on push, 404 on nonexistent vehicle) - **TestContactPersons** (5 tests): Add person, remove person, 404 cases, detail includes persons
- TestMobileDeStatus: 3 tests (200 with no listing, 200 with synced listing, 404 on nonexistent) - **TestContactRBAC** (12 tests): Auth required, admin+verkaeufer write, verkaeufer read, all endpoints RBAC
- **TestContactServiceDirect** (16 tests): Direct service-level coverage for all functions
#### test_mobilede.py (18 tests)
- TestFieldMapping: 8 tests (LKW, Baumaschine, PKW, Stapler, Transporter, no optional fields, lkw_type with prefix, unknown lkw_type fallback)
- TestPushListing: 3 tests (success, HTTP error, request error)
- TestUpdateListing: 2 tests (success, no ad_id)
- TestDeleteListing: 2 tests (success, no ad_id)
- TestGetListingStatus: 2 tests (returns latest, returns none)
- TestRetryFailedListing: 2 tests (succeeds within max retries, exceeds max retries)
#### test_vehicles_extra.py (28 tests)
- TestVehicleServiceDirect: 21 tests (list empty, pagination, sort asc, invalid sort, min/max price, search by fin/location, get by fin, create, duplicate fin, update, not found, duplicate fin update, same fin, soft delete, not found, get by id not found, excludes deleted)
- TestRouterAdditionalPaths: 7 tests (all filters combined, verkaeufer allowed, fin duplicate 409, empty result, invalid UUID 422, push failure 202, status after failed push)
## Frontend Tests ## Frontend Tests
**Command**: `cd frontend && npx vitest run tests/vehicles.test.tsx` **Command:** `cd frontend && npx vitest run tests/contacts.test.tsx`
**Result**: 16 passed in 2.21s **Result:** 20 passed, 0 failed
### Test Categories ### Test Categories
#### VehicleList (5 tests) - **ContactList** (5 tests): Renders table with search/filter, displays contacts, new contact button, error handling, loading state
- Renders vehicle list with filters and table - **ContactForm** (7 tests): All fields rendered, EU/Inland toggle, defaults to DE, VAT ID validation, valid submission, required fields
- Displays vehicles in table after loading - **validateVatIdFormat** (8 tests): DE/AT/NL valid, invalid DE, non-EU country, lowercase, spaces
- Shows pagination when total > page_size
- Shows error message on API failure
- Calls listVehicles with type filter when changed
#### VehicleForm (4 tests)
- Renders form with all required fields
- Shows validation errors for empty required fields
- Shows error for FIN not 17 characters
- Calls createVehicle on submit with valid data
#### VehicleDetail (3 tests)
- Renders vehicle details after loading
- Shows mobile.de status section
- Shows error message on API failure
#### MobileDeStatus (4 tests)
- Renders with pending status when no listing exists
- Shows synced status after successful push
- Shows error log when sync failed
- Calls pushToMobileDe when push button is clicked
## Smoke Test ## Smoke Test
- Backend: All API endpoints tested via HTTPX ASGI transport with real PostgreSQL test DB - Backend: All 7 API endpoints tested via HTTPX ASGI transport with real PostgreSQL test database
- Frontend: All components tested with React Testing Library and mocked API calls - Frontend: Component rendering tested with jsdom, API calls mocked
- mobile.de: All HTTP calls mocked with httpx.AsyncClient patches, no real API calls made - RBAC: admin and verkaeufer roles tested for all write endpoints
- Auth: All vehicle endpoints require JWT Bearer token (verified with 401 test) - Soft-delete: Verified deleted contacts return 404 and don't appear in list
- Soft-delete: Verified deleted_at is set and vehicle excluded from subsequent queries - VAT ID validation: Both backend (Pydantic field_validator) and frontend (validateVatIdFormat) tested
## Acceptance Criteria Verification ## Acceptance Criteria Verification
| Criterion | Status | Evidence | | Criterion | Status | Evidence |
|-----------|--------|----------| |-----------|--------|----------|
| GET /api/v1/vehicles → 200 + paginated list | ✅ | test_list_vehicles_returns_200_with_pagination | | GET /api/v1/contacts → 200 + paginated list | ✅ | test_list_contacts_returns_200_with_pagination |
| GET /api/v1/vehicles?type=lkw&availability=available → 200 + filtered | ✅ | test_list_vehicles_filter_by_type, test_list_vehicles_filter_by_availability | | GET /api/v1/contacts?role=kaeufer → filtered (beide included) | ✅ | test_list_contacts_filter_by_role_kaeufer_includes_beide |
| GET /api/v1/vehicles?sort=-created_at → 200 + sorted | ✅ | test_list_vehicles_sort_descending | | GET /api/v1/contacts?is_eu=true → EU contacts only | ✅ | test_list_contacts_filter_is_eu_true |
| GET /api/v1/vehicles/:id → 200 + detail, nonexistent → 404 | ✅ | test_get_vehicle_returns_200, test_get_vehicle_nonexistent_returns_404 | | GET /api/v1/contacts?search=mueller → matching contacts | ✅ | test_list_contacts_search_by_company_name |
| POST /api/v1/vehicles valid → 201, missing make → 422 | ✅ | test_create_vehicle_returns_201, test_create_vehicle_missing_make_returns_422 | | GET /api/v1/contacts/:id → detail with contact persons | ✅ | test_get_contact_returns_200_with_detail |
| PUT /api/v1/vehicles/:id → 200 + updated | ✅ | test_update_vehicle_returns_200 | | GET /api/v1/contacts/:nonexistent → 404 | ✅ | test_get_contact_nonexistent_returns_404 |
| DELETE /api/v1/vehicles/:id → 200 + deleted_at set | ✅ | test_delete_vehicle_returns_200_with_deleted_at | | POST /api/v1/contacts valid → 201 | ✅ | test_create_contact_returns_201 |
| POST /api/v1/vehicles/:id/mobile-de/push → 202 (async) | ✅ | test_push_returns_202 | | POST /api/v1/contacts invalid vat_id → 422 | ✅ | test_create_contact_with_invalid_vat_id_returns_422 |
| GET /api/v1/vehicles/:id/mobile-de/status → 200 + sync info | ✅ | test_status_returns_200_with_synced_listing | | PUT /api/v1/contacts/:id → 200 + updated | ✅ | test_update_contact_returns_200 |
| mobile.de push sends correct Ad format (mocked) | ✅ | test_map_fields_basic_lkw, test_push_listing_success | | DELETE /api/v1/contacts/:id → 200 (soft delete) | ✅ | test_delete_contact_returns_200 |
| Frontend Vehicle List renders table with filter+pagination | ✅ | VehicleList test suite (5 tests) | | Frontend Contact List renders table with search + filter | ✅ | renders contact list with search and filter controls |
| Frontend Vehicle Detail shows all fields + mobile.de status | ✅ | VehicleDetail test suite (3 tests) | | Frontend Contact Form validates USt-IdNr. format | ✅ | validates VAT ID format and shows error for invalid DE VAT |
| Frontend Create Form validates required fields | ✅ | VehicleForm test suite (4 tests) | | Frontend EU/Inland toggle changes required fields | ✅ | switches to EU mode and shows VAT ID hint when EU is selected |
| pytest coverage >= 80% vehicle module | ✅ | 82% total coverage | | pytest coverage >= 80% contact module | ✅ | 91% total coverage |