QA Test Framework: Architecture¶

┌─────────────────────────────────────────────────┐
│         Presentation Layer (BDD)                │
│  - Feature files (.feature)                     │
│  - Gherkin scenarios                            │
└─────────────────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────────────────┐
│         Step Definition Layer                    │
│  - Step implementations                          │
│  - Context management                            │
│  - Orchestration logic                           │
└─────────────────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────────────────┐
│         Business Logic Layer                     │
│  - ImportConfig (document builder)               │
│  - AccountingDocumentBase (expectations)         │
│  - Helper functions                              │
└─────────────────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────────────────┐
│         Model Layer                              │
│  - Pydantic models                               │
│  - Enumerations                                  │
│  - Data validation                               │
└─────────────────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────────────────┐
│         Service Layer                            │
│  - HTTP clients (Controller)                     │
│  - Authentication (AuthController)               │
│  - External integrations                         │
└─────────────────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────────────────┐
│         External Systems                         │
│  - Accounting Service API                        │
│  - Keycloak                                      │
│  - Database (indirect)                           │
└─────────────────────────────────────────────────┘

class TranslationFactory:
  """Factory for creating translation models based on country"""

  @classmethod
  def get_translation_model(cls, vat_country: str) -> Type[BaseModel] | None:
    """
    Returns appropriate translation model for country.

    Args:
        vat_country: ISO country code (AE, SA, OM)

    Returns:
        Translation model class or None
    """
    match vat_country:
      case "AE" | "SA" | "OM":
        return ArabicInvoiceModel
      case _:
        return None

  @classmethod
  def get_provider(cls, country_code: str) -> TranslationProvider:
    """Get translation provider for specific country"""
    return cls._providers.get(country_code, NoTranslationProvider())

class ImportConfig:
  """Builder for document import requests"""

  def __init__(self, init_data: dict):
    """Initialize with table data from feature file"""
    self._init = ImportConfigInitModel(**init_data)
    self._setup_derived_attributes()

  def build_import_document(self, user_data: User) -> dict:
    """
    Build complete import document structure.

    Returns:
        Complete document JSON ready for API
    """
    self.document_body = {
      "init": self._init_db_data,
      "invoice": [],
      "organization": self._organization_data,
    }

    # Build main invoice item
    main_item = self._create_invoice_item()

    # Add metadata if present
    if self._init.payment_info:
      main_item["Metadata"] = self._payment_info

    self.document_body["invoice"].append(main_item)

    # Add additional items (points, discounts, etc.)
    self._add_additional_items()

    return self.document_body

class ItemType(Enum):
  """Item type with platform-specific strategies"""

  def account_number(self, db_setting: DatabaseSetting) -> str:
    """Get account number based on platform strategy"""
    match db_setting.platform:
      case Platform.HELIOS:
        return self._helios_account_number()
      case Platform.XERO:
        return self._xero_account_number()
      case Platform.NETSUITE:
        return self._netsuite_account_number()
      case _:
        raise NotImplementedError(f"Platform {db_setting.platform}")

  def _helios_account_number(self) -> str:
    """Helios-specific account number logic"""
    return CisloUcetPol.get_helios_account_number(self)

  def _xero_account_number(self) -> str:
    """Xero-specific account number logic"""
    return self.value.xero_type

  def _netsuite_account_number(self) -> str:
    """NetSuite-specific account number logic"""
    return self.value.code

class MixedCaseModel(BaseModel):
  """
  Adapter for Pydantic models with PascalCase aliases.
  Converts Python snake_case to JSON PascalCase.
  """

  model_config = ConfigDict(
    alias_generator=to_pascal,  # Convert field names
    populate_by_name=True,  # Accept both formats
    str_strip_whitespace=True,  # Clean strings
    validate_assignment=True,  # Validate on change
  )

  def compare(self, other: "MixedCaseModel", **kwargs) -> dict:
    """Compare two models using DeepDiff"""
    return DeepDiff(
      self.model_dump(by_alias=True),
      other.model_dump(by_alias=True),
      **kwargs
    )

class AuthController:
  """Repository for user data and authentication"""

  def __init__(self, users: dict, kc_base_url: str, default_client: str):
    self.users = users  # User repository
    self._kc_base_url = kc_base_url
    self._default_client = default_client
    self._token_cache: dict[str, Token] = {}

  def get_token(self, user_id: str, client_id: str) -> Token:
    """
    Retrieve token from cache or generate new one.

    Repository pattern: Abstracts token storage/retrieval.
    """
    cache_key = f"{user_id}:{client_id}"

    if cache_key in self._token_cache:
      token = self._token_cache[cache_key]
      if not token.is_expired():
        return token

    # Generate new token
    token = self._fetch_token_from_keycloak(user_id, client_id)
    self._token_cache[cache_key] = token
    return token

def before_all(context: Context) -> None:
  """
  Initialize global test context before any tests run.

  Responsibilities:
  - Load configuration
  - Initialize logging
  - Create service controllers
  - Setup authentication
  """
  context.settings = AccountingServiceTestConfig()
  configure_logging(context.settings.log_level)

  auth_controller = AuthController(
    users=context.settings.users,
    kc_base_url=StaticConfig.KEYCLOAK_BASE_URL,
    default_client=context.settings.users.default_client,
  )

  context.controller = Controller(
    auth_controller=auth_controller,
    base_url=context.settings.base_url,
  )


def before_feature(context: Context, feature: Feature) -> None:
  """Initialize feature-level context"""
  context.users = context.controller.auth.users


def before_scenario(context: Context, scenario: Scenario) -> None:
  """Initialize scenario-level context"""
  context.sent_import_data = None
  context.import_config = None
  context.imported_documents = []
  context.gcs_blobs_to_delete = []


def after_scenario(context: Context, scenario: Scenario) -> None:
  """Cleanup after scenario"""
  for bucket_name, blob_name in getattr(context, "gcs_blobs_to_delete", []):
    delete_gcs_blob(bucket_name, blob_name)

class StaticConfig:
  """Static configuration from environment variables"""

  KEYCLOAK_BASE_URL = os.getenv("KEYCLOAK_BASE_URL")
  LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")

  @classmethod
  def validate(cls) -> None:
    """Validate all required config is present"""
    required = ["KEYCLOAK_BASE_URL"]
    missing = [k for k in required if not getattr(cls, k)]
    if missing:
      raise ConfigurationError(f"Missing: {missing}")


class AccountingServiceTestConfig:
  """Runtime configuration with validation"""

  def __init__(self):
    self.base_url = os.getenv("BASE_URL")
    self.timeout = int(os.getenv("TIMEOUT", "30"))
    self.log_level = os.getenv("LOG_LEVEL", "INFO")

    # Load users from secrets manager or config
    self.users = self._load_users()

    # Validate configuration
    self._validate()

  def _validate(self) -> None:
    """Validate configuration is complete"""
    if not self.base_url:
      raise ConfigurationError("BASE_URL not set")

def configure_logging(log_level: str = "INFO") -> None:
  """
  Configure logging with custom format and handlers.

  Logger hierarchy:
  - root: Catches all
  - steps_logger: Step definitions
  - http_logger: HTTP requests
  - validation_logger: Validation
  """
  logging.basicConfig(
    level=getattr(logging, log_level.upper()),
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
      logging.FileHandler('logs/test_execution.log'),
      logging.StreamHandler(sys.stdout)
    ]
  )

  # Configure specific loggers
  logging.getLogger('urllib3').setLevel(logging.WARNING)
  logging.getLogger('requests').setLevel(logging.WARNING)


# Specialized loggers
steps_logger = logging.getLogger('steps')
http_logger = logging.getLogger('http')
validation_logger = logging.getLogger('validation')

BaseModel (Pydantic)
│
├── MixedCaseModel (Custom base)
│   ├── ImportConfigInitModel
│   ├── Invoice
│   ├── JsonDocument
│   ├── JsonDocumentItem
│   ├── JsonDocumentResponse
│   ├── OandaJsonDocumentResponse
│   └── InvoiceModel
│
├── Item (Document item representation)
├── Organization (Organization data)
└── User (User data)

class InvoiceModel(MixedCaseModel):
  """HTML document validation model"""

  @classmethod
  def from_html(cls, html_dict: dict, document_body: dict) -> Self:
    """Create from parsed HTML"""
    return cls(
      document_id=html_dict.get("document-id"),
      invoice_value=html_dict.get("invoice-number"),
      # ... extract and transform fields
    )

  @classmethod
  def from_document_base(cls, document_base: AccountingDocumentBase) -> Self:
    """Create from expected values"""
    payment_info = MetadataModel.build_expected_data(document_base)
    organization_info = OrganizationInfoModel.build_expected_data(document_base)
    invoice_dto = InvoiceModel.build_expected_data(document_base)

    return cls(**{
      **invoice_dto.model_dump(),
      **payment_info.model_dump(),
      **organization_info.model_dump()
    })

class JsonDocumentResponse(MixedCaseModel):
  """Response model with validation"""

  id: str = Field(min_length=8, max_length=12)
  document: JsonDocument
  documentItems: list[JsonDocumentItem] = Field(min_length=1)

  model_config = ConfigDict(
    str_strip_whitespace=True,  # Clean strings
    validate_assignment=True,  # Validate on change
    extra='forbid'  # Reject unknown fields
  )

  @field_validator('id')
  @classmethod
  def validate_document_id(cls, v: str) -> str:
    """Custom validation for document ID format"""
    if not v.startswith(('01', '02', '90', '80', '70')):
      raise ValueError(f"Invalid document ID prefix: {v}")
    return v

class Controller:
  """
  Main HTTP client for accounting service.

  Architecture:
  - Handles authentication via AuthController
  - Provides high-level API methods
  - Manages request/response lifecycle
  - Implements retry logic
  """

  def __init__(self, auth_controller: AuthController, base_url: str):
    self.auth = auth_controller
    self.base_url = base_url
    self.timeout = 30
    self.session = requests.Session()

  def __getitem__(self, key: tuple[str, str]) -> "Controller":
    """
    Indexer pattern for setting auth context.

    Usage:
        controller = context.controller["CLIENT_NBO", "service"]
        controller.import_document(body)
    """
    user_id, client_id = key
    self._current_user = user_id
    self._current_client = client_id
    return self

  def _get_headers(self) -> dict:
    """Build headers with authentication"""
    token = self.auth.get_token(self._current_user, self._current_client)
    return {
      "Authorization": f"Bearer {token.access_token}",
      "Content-Type": "application/json",
      "Accept": "application/json"
    }

  def _request(
    self,
    method: str,
    endpoint: str,
    **kwargs
  ) -> requests.Response:
    """
    Internal request method with error handling.

    Handles:
    - Timeout
    - Connection errors
    - HTTP errors
    - Logging
    """
    url = f"{self.base_url}{endpoint}"

    try:
      http_logger.debug(f"{method} {url}")
      response = self.session.request(
        method=method,
        url=url,
        headers=self._get_headers(),
        timeout=self.timeout,
        **kwargs
      )
      http_logger.debug(f"Response: {response.status_code}")
      return response

    except requests.Timeout:
      http_logger.error(f"Timeout after {self.timeout}s")
      raise
    except requests.ConnectionError as e:
      http_logger.error(f"Connection error: {e}")
      raise

  def import_document(self, body: dict) -> requests.Response:
    """Import document to accounting system"""
    return self._request(
      "POST",
      "/v1/service/import/document/id",
      json=body
    )

  def get_html_document(self, cid: int, document_id: str) -> requests.Response:
    """Retrieve HTML document"""
    return self._request(
      "GET",
      f"/v1/user/{cid}/document/{document_id}/html"
    )

class AuthController:
  """
  Manages authentication tokens with caching.

  Features:
  - Token caching
  - Automatic refresh
  - Multiple client support
  """

  def __init__(self, users: dict, kc_base_url: str, default_client: str):
    self.users = users
    self._kc_base_url = kc_base_url
    self._default_client = default_client
    self._token_cache: dict[str, Token] = {}
    self._cache_lock = threading.Lock()

  def get_token(self, user_id: str, client_id: str) -> Token:
    """
    Get token with thread-safe caching.

    Flow:
    1. Check cache
    2. Validate expiry
    3. Return cached or fetch new
    """
    cache_key = f"{user_id}:{client_id}"

    with self._cache_lock:
      if cache_key in self._token_cache:
        token = self._token_cache[cache_key]
        if not token.is_expired():
          return token

      # Fetch new token
      token = self._fetch_token_from_keycloak(user_id, client_id)
      self._token_cache[cache_key] = token
      return token

  def _fetch_token_from_keycloak(
    self,
    user_id: str,
    client_id: str
  ) -> Token:
    """Fetch token from Keycloak"""
    user = self.users[user_id]

    response = requests.post(
      f"{self._kc_base_url}/realms/{self._realm}/protocol/openid-connect/token",
      data={
        "grant_type": "password",
        "client_id": client_id,
        "username": user.username,
        "password": user.password,
      }
    )

    if response.status_code != 200:
      raise AuthenticationError(f"Failed to get token: {response.text}")

    token_data = response.json()
    return Token(
      access_token=token_data["access_token"],
      expires_in=token_data["expires_in"],
      refresh_token=token_data.get("refresh_token"),
      fetched_at=datetime.now()
    )

        ┌─────────────┐
        │   E2E Tests │  (Feature files)
        │   (Slow)    │  Full flow validation
        └─────────────┘
             ▲
        ┌────────────────┐
        │ Integration    │  (Step definitions + HTTP)
        │ Tests (Medium) │  API interaction
        └────────────────┘
             ▲
        ┌────────────────────┐
        │   Unit Tests       │  (Helpers, Models)
        │   (Fast)           │  Business logic
        └────────────────────┘

test_data/
├── payment_info_special_tip_a.json
├── payment_info_special_tip_b.json
├── payment_info_points.json
└── payment_info_challenge.json

class Company(Enum):
  """Company data as enum"""
  FTMO_EVAL_GLOBAL = CompanyField(
    item_id=5,
    full_name="FTMO Evaluation Global s.r.o.",
    address="Purkyňova 2121/3",
    # ... all company data
  )

def create_test_user(
  user_id: str,
  database: DatabaseSetting,
  country: Country = Country.CZ
) -> User:
  """Factory for creating test users"""
  return User(
    cid=generate_cid(),
    username=f"test_{user_id}",
    database=database,
    country=country,
  )

# enums/enums_db_setting.py
class Platform(Enum):
  HELIOS = auto()
  XERO = auto()
  NETSUITE = auto()
  NEW_PLATFORM = auto()  # Add new platform

# enums/enums_item_types.py
class ItemType(Enum):
  def item_id(self, db_setting: DatabaseSetting) -> str:
    match db_setting.platform:
      case Platform.HELIOS:
        return self.value.helios_type
      case Platform.XERO:
        return self.value.xero_type
      case Platform.NETSUITE:
        return self.value.code
      case Platform.NEW_PLATFORM:
        return self.value.new_platform_type  # Add mapping

# enums/enums_db_setting.py
class DatabaseSetting(Enum):
  NEW_DB = DatabaseConfig(
    db_name="NewPlatformDB",
    platform=Platform.NEW_PLATFORM,
    company=Company.NEW_COMPANY,
    # ... config
  )

@cached_property
def total_vat_czk_money(self) -> Money:
  """Expensive calculation - cached"""
  if not self.show_czk:
    return Money(0, Currency.CZK)

  # Complex calculation
  czk_amount = self._calculate_vat_in_czk()
  return Money(czk_amount, Currency.CZK)

class Controller:
  def __init__(self, ...):
    self.session = requests.Session()  # Connection pool
    self.session.mount('https://', HTTPAdapter(
      pool_connections=10,
      pool_maxsize=20,
      max_retries=3
    ))

# Run tests in parallel
behave --processes 4 --parallel-element scenario

# ❌ Wrong
password = "mypassword123"

# ✅ Correct
password = os.getenv("USER_PASSWORD")

# ✅ Better - use secrets manager
password = secrets_manager.get_secret("qa/user/password")

class Token:
  """Secure token with expiry"""

  def __init__(self, access_token: str, expires_in: int, ...):
    self._access_token = access_token  # Private
    self._expires_at = datetime.now() + timedelta(seconds=expires_in)

  @property
  def access_token(self) -> str:
    """Read-only access to token"""
    if self.is_expired():
      raise TokenExpiredError("Token has expired")
    return self._access_token

  def is_expired(self) -> bool:
    """Check if token is expired (with buffer)"""
    buffer = timedelta(seconds=30)
    return datetime.now() >= (self._expires_at - buffer)

  def __repr__(self) -> str:
    """Don't expose token in repr"""
    return f"Token(expires_at={self._expires_at})"

def log_request(url: str, headers: dict, body: dict):
  """Log request with sensitive data masked"""
  safe_headers = {
    k: "***REDACTED***" if k.lower() in ["authorization", "api-key"] else v
    for k, v in headers.items()
  }

  safe_body = mask_sensitive_fields(body, ["password", "secret", "token"])

  http_logger.debug(f"Request: {url}")
  http_logger.debug(f"Headers: {safe_headers}")
  http_logger.debug(f"Body: {json.dumps(safe_body)}")