Testing Data Models: A Systematic Approach to Finding Edge Cases

from pydantic import BaseModel
from typing import Optional
from enum import Enum

class SampleType(str, Enum):
    CONTROL = "control"
    EXPERIMENTAL = "experimental"
    CALIBRATION = "calibration"

class Sample(BaseModel):
    sample_id: str
    experiment_id: str
    concentration_mM: Optional[float] = None
    sample_type: SampleType = SampleType.EXPERIMENTAL
    is_validated: bool = True

class SpectroscopyReading(BaseModel):
    # Required fields
    reading_id: str
    instrument_id: str

    # Optional fields
    wavelength_nm: Optional[float] = None
    temperature_K: Optional[float] = None
    pressure_atm: Optional[float] = None
    notes: Optional[str] = None

    # Enums
    sample_type: SampleType = SampleType.EXPERIMENTAL  # 3 values
    status: ReadingStatus = ReadingStatus.PENDING  # 4 values: PENDING, VALIDATED, FLAGGED, REJECTED
    instrument_mode: InstrumentMode = InstrumentMode.STANDARD  # 5 values: STANDARD, HIGH_RES, FAST, CALIBRATION, DIAGNOSTIC

    # Booleans
    is_validated: bool = True
    requires_review: bool = False
    is_replicate: bool = False

import plotly.graph_objects as go
import numpy as np

# Use consistent x-axis (0-10 fields) for clear comparison
num_fields = list(range(0, 11))

# Scenario 1: Boolean/Optional fields only (2^n growth)
combinations_binary = [2**n for n in num_fields]

# Scenario 2: Enum fields only (avg 3 values each: 3^n growth)
combinations_enum = [3**n for n in num_fields]

# Scenario 3: Mixed realistic model
# Alternating: booleans (2x) and small enums (3x)
combinations_mixed = [1]
for i in range(1, 11):
    multiplier = 2 if i % 2 == 1 else 3
    combinations_mixed.append(combinations_mixed[-1] * multiplier)

fig = go.Figure()

# Plot in order of growth rate for visual clarity
fig.add_trace(go.Scatter(
    x=num_fields,
    y=combinations_binary,
    mode='lines+markers',
    name='Boolean/Optional (2ⁿ)',
    line=dict(color='#2BCDC1', width=3),
    marker=dict(size=8)
))

fig.add_trace(go.Scatter(
    x=num_fields,
    y=combinations_mixed,
    mode='lines+markers',
    name='Mixed Model (~2.4ⁿ)',
    line=dict(color='#FFB347', width=3),
    marker=dict(size=8)
))

fig.add_trace(go.Scatter(
    x=num_fields,
    y=combinations_enum,
    mode='lines+markers',
    name='Enum Fields (3ⁿ)',
    line=dict(color='#F66095', width=3),
    marker=dict(size=8)
))

# Add reference lines for context
fig.add_hline(y=100, line_dash="dash", line_color="gray",
              annotation_text="100 tests", annotation_position="bottom right")
fig.add_hline(y=10000, line_dash="dash", line_color="red",
              annotation_text="10,000 tests", annotation_position="bottom right")

fig.update_layout(
    title='Structural Combinations vs Number of Fields',
    xaxis_title='Number of Fields Added',
    yaxis_title='Test Combinations (log scale)',
    yaxis_type='log',
    yaxis=dict(range=[0, 5]),  # 10^0 to 10^5 for cleaner view
    xaxis=dict(dtick=1),  # Show every integer on x-axis
    template='plotly_dark',
    legend=dict(x=0.02, y=0.98, bgcolor='rgba(0,0,0,0.5)'),
    font=dict(size=12),
    hovermode='x unified'
)

from orgutils import plotly_figure_to_json, plotly_tight_layout
plotly_tight_layout(fig)
plotly_figure_to_json(fig, "../static/dm_combinatorial_growth.json")

import plotly.graph_objects as go
import numpy as np

# Model sizes from 3 to 10 fields (more reasonable range)
model_sizes = list(range(3, 11))

# Base structural combinations (doubling for each field as approximation)
base_structural = [12 * (2 ** (n - 3)) for n in model_sizes]

# Value variations: 2, 3, or 4 edge cases per field
value_multipliers = [2, 3, 4]
colors = ['#2BCDC1', '#FFB347', '#F66095']
labels = ['2 values/field (min/max)', '3 values/field (+boundary)', '4 values/field (+zero)']

fig = go.Figure()

for mult, color, label in zip(value_multipliers, colors, labels):
    # Total = structural * (value_variations ^ num_fields)
    total_tests = [base * (mult ** size) for base, size in zip(base_structural, model_sizes)]
    fig.add_trace(go.Scatter(
        x=model_sizes,
        y=total_tests,
        mode='lines+markers',
        name=label,
        line=dict(color=color, width=3),
        marker=dict(size=8)
    ))

# Add meaningful reference lines
fig.add_hline(y=1000, line_dash="dot", line_color="gray",
              annotation_text="1K tests", annotation_position="bottom right")
fig.add_hline(y=1e6, line_dash="dash", line_color="orange",
              annotation_text="1M tests", annotation_position="bottom right")
fig.add_hline(y=1e9, line_dash="dash", line_color="red",
              annotation_text="1B tests", annotation_position="bottom right")

fig.update_layout(
    title='Total Test Space: Structure × Value Combinations',
    xaxis_title='Number of Fields in Model',
    yaxis_title='Total Test Cases (log scale)',
    yaxis_type='log',
    yaxis=dict(range=[2, 11]),  # 10^2 to 10^11
    xaxis=dict(dtick=1),
    template='plotly_dark',
    legend=dict(x=0.02, y=0.98, bgcolor='rgba(0,0,0,0.5)'),
    font=dict(size=12),
    hovermode='x unified'
)

from orgutils import plotly_figure_to_json, plotly_tight_layout
plotly_tight_layout(fig)
plotly_figure_to_json(fig, "../static/dm_total_test_space.json")

from polyfactory.factories.pydantic_factory import ModelFactory
from pydantic import BaseModel
from typing import Optional
from enum import Enum

class SampleType(str, Enum):
    CONTROL = "control"
    EXPERIMENTAL = "experimental"
    CALIBRATION = "calibration"

class Sample(BaseModel):
    sample_id: str
    experiment_id: str
    concentration_mM: Optional[float] = None
    sample_type: SampleType = SampleType.EXPERIMENTAL
    is_validated: bool = True

class SampleFactory(ModelFactory):
    __model__ = Sample

# Generate a random valid sample
sample = SampleFactory.build()
print(sample)
# Sample(sample_id='xKjP2mQ', experiment_id='exp-001', concentration_mM=42.5, sample_type='experimental', is_validated=True)

# Override specific fields
control_sample = SampleFactory.build(sample_type=SampleType.CONTROL, is_validated=True)

sample_id='ETtjoDOBbxvWWUsLdaHi' experiment_id='pJRXkKZrahIBvHFnrvCh' concentration_mM=-193113322441.84 sample_type=<SampleType.CALIBRATION: 'calibration'> is_validated=False

# Generate instances covering all structural variations
for sample in SampleFactory.coverage():
    print(f"type={sample.sample_type}, conc={'set' if sample.concentration_mM else 'None'}, valid={sample.is_validated}")

type=SampleType.CONTROL, conc=set, valid=True
type=SampleType.EXPERIMENTAL, conc=None, valid=True
type=SampleType.CALIBRATION, conc=set, valid=False

def determine_priority(sample: Sample) -> str:
    """Determine analysis priority based on sample type and validation status."""

    # Calibration samples are always high priority
    if sample.sample_type == SampleType.CALIBRATION:
        return "high"

    # Unvalidated samples need review first
    if not sample.is_validated:
        raise ValueError("Sample must be validated before analysis")

    # Control samples with known concentration get medium priority
    if sample.sample_type == SampleType.CONTROL and sample.concentration_mM is not None:
        return "medium"

    return "normal"

import pytest

def test_priority_all_sample_variations():
    """Test priority determination across all sample variations."""
    results = []
    for sample in SampleFactory.coverage():
        if not sample.is_validated:
            try:
                determine_priority(sample)
                results.append(f"FAIL: {sample.sample_type.value}, validated={sample.is_validated} - expected ValueError")
            except ValueError:
                results.append(f"PASS: {sample.sample_type.value}, validated={sample.is_validated} - correctly raised ValueError")
        elif sample.sample_type == SampleType.CALIBRATION:
            priority = determine_priority(sample)
            if priority == "high":
                results.append(f"PASS: {sample.sample_type.value}, validated={sample.is_validated} -> {priority}")
            else:
                results.append(f"FAIL: {sample.sample_type.value} expected 'high', got '{priority}'")
        else:
            priority = determine_priority(sample)
            if priority in ["high", "medium", "normal"]:
                results.append(f"PASS: {sample.sample_type.value}, validated={sample.is_validated} -> {priority}")
            else:
                results.append(f"FAIL: {sample.sample_type.value} got invalid priority '{priority}'")
    return results

# Run the test and display results
print("Testing priority determination across all sample variations:")
print("-" * 60)
for result in test_priority_all_sample_variations():
    print(result)
print("-" * 60)
print(f"All {len(list(SampleFactory.coverage()))} variations tested!")
assert False  # Force output display in org-mode

import pytest
from typing import Type, Iterator, TypeVar
from pydantic import BaseModel
from polyfactory.factories.pydantic_factory import ModelFactory

T = TypeVar("T", bound=BaseModel)

def create_factory(model: Type[T]) -> Type[ModelFactory[T]]:
    """Dynamically create a factory for any Pydantic model."""
    return type(f"{model.__name__}Factory", (ModelFactory,), {"__model__": model})

@pytest.fixture
def model_coverage(request) -> Iterator[BaseModel]:
    """
    Reusable fixture that yields all structural variations of a model.

    Usage:
        @pytest.mark.parametrize("model_class", [Sample, Measurement, Experiment])
        def test_serialization(model_coverage, model_class):
            for instance in model_coverage:
                assert instance.model_dump_json()
    """
    model_class = request.param
    factory = create_factory(model_class)
    yield from factory.coverage()

# Now testing ANY model is trivial:
@pytest.mark.parametrize("model_class", [Sample, SpectroscopyReading, Experiment])
def test_all_models_serialize(model_class):
    """Every model variation must serialize to JSON."""
    factory = create_factory(model_class)
    for instance in factory.coverage():
        json_str = instance.model_dump_json()
        restored = model_class.model_validate_json(json_str)
        assert restored == instance

from hypothesis import given, strategies as st, settings

@given(st.integers())
@settings(max_examples=10)  # Limit for demo
def test_absolute_value_is_non_negative(n):
    """Property: absolute value is always >= 0"""
    assert abs(n) >= 0

@given(st.text())
@settings(max_examples=10)  # Limit for demo
def test_string_reversal_is_reversible(s):
    """Property: reversing twice gives original"""
    assert s[::-1][::-1] == s

# Run the tests and show output
print("Running Hypothesis tests:")
print("-" * 60)
try:
    test_absolute_value_is_non_negative()
    print("PASS: test_absolute_value_is_non_negative - all generated integers passed")
except AssertionError as e:
    print(f"FAIL: test_absolute_value_is_non_negative - {e}")

try:
    test_string_reversal_is_reversible()
    print("PASS: test_string_reversal_is_reversible - all generated strings passed")
except AssertionError as e:
    print(f"FAIL: test_string_reversal_is_reversible - {e}")
print("-" * 60)
assert False  # Force output display

@given(st.text())
def test_sample_notes_field(notes: str):
    """What could go wrong with a simple notes field?"""
    sample = Sample(
        sample_id="test-001",
        experiment_id="exp-001",
        notes=notes  # Oh no.
    )
    process_sample(sample)

from hypothesis import given, strategies as st
from hypothesis import settings

@given(st.from_type(Sample))
@settings(max_examples=20)  # Reduced for demo output
def test_sample_serialization_roundtrip(sample: Sample):
    """Property: serializing and deserializing preserves data"""
    json_str = sample.model_dump_json()
    restored = Sample.model_validate_json(json_str)
    assert restored == sample

# Run and show output
print("Testing Sample serialization roundtrip with Hypothesis:")
print("-" * 60)
try:
    test_sample_serialization_roundtrip()
    print("PASS: All 20 generated Sample instances serialized correctly")
except AssertionError as e:
    print(f"FAIL: {e}")
except Exception as e:
    print(f"ERROR: {type(e).__name__}: {e}")
print("-" * 60)
assert False  # Force output display

from hypothesis import given, strategies as st, assume

# Strategy for wavelengths (must be positive, typically 200-1100nm for UV-Vis)
valid_wavelength = st.floats(min_value=200.0, max_value=1100.0, allow_nan=False)

# Strategy for temperature (above absolute zero, below plasma)
valid_temperature = st.floats(min_value=0.001, max_value=10000.0, allow_nan=False)

# Strategy for concentration (non-negative, physically reasonable)
valid_concentration = st.one_of(
    st.none(),
    st.floats(min_value=0.0, max_value=1000.0, allow_nan=False)  # millimolar
)

# Strategy for pressure (vacuum to high pressure, in atmospheres)
valid_pressure = st.floats(min_value=0.0, max_value=1000.0, allow_nan=False)

# Composite strategy with inter-field constraints
@st.composite
def spectroscopy_reading_strategy(draw):
    """Generate physically plausible spectroscopy readings."""
    wavelength = draw(valid_wavelength)
    temperature = draw(valid_temperature)
    pressure = draw(valid_pressure)

    # Domain constraint: at very low pressure, temperature readings are unreliable
    # (this is a real thing in vacuum spectroscopy!)
    if pressure < 0.01:
        assume(temperature > 100)  # Skip implausible combinations

    return SpectroscopyReading(
        reading_id=draw(st.text(min_size=1, max_size=50).filter(str.strip)),
        instrument_id=draw(st.sampled_from(["UV-1800", "FTIR-4600", "Raman-532"])),
        wavelength_nm=wavelength,
        temperature_K=temperature,
        pressure_atm=pressure,
        sample_type=draw(st.sampled_from(SampleType)),
        is_validated=draw(st.booleans())
    )

@given(spectroscopy_reading_strategy())
def test_reading_within_physical_bounds(reading: SpectroscopyReading):
    """Property: all readings must be physically plausible"""
    if reading.wavelength_nm is not None:
        assert reading.wavelength_nm > 0, "Negative wavelength is not a thing"
    if reading.temperature_K is not None:
        assert reading.temperature_K > 0, "Below absolute zero? Bold claim."

import plotly.graph_objects as go
from plotly.subplots import make_subplots
import numpy as np
from hypothesis import strategies as st, settings, Phase
from hypothesis.strategies import SearchStrategy

# =============================================================================
# REAL Hypothesis sampling vs naive uniform random
# We use Hypothesis's draw() mechanism to collect actual generated values
# =============================================================================

# Collect samples from Hypothesis's floats strategy
# Hypothesis uses a combination of: boundary values, special floats, and random exploration
hypothesis_wavelengths = []
wavelength_strategy = st.floats(min_value=200.0, max_value=1100.0, allow_nan=False, allow_infinity=False)

# Use find() with a condition that always fails to force Hypothesis to explore the space
# This collects the actual values Hypothesis would use in testing
from hypothesis import find, Verbosity
from hypothesis.errors import NoSuchExample

for seed in range(500):
    try:
        # Use example() which gives us actual Hypothesis-generated values
        val = wavelength_strategy.example()
        hypothesis_wavelengths.append(val)
    except Exception:
        pass

# Naive uniform random for comparison
np.random.seed(42)
naive_wavelengths = np.random.uniform(200, 1100, 500)

# Print some statistics to show the difference
print("Sampling Comparison (Wavelength 200-1100nm):")
print("-" * 50)
print(f"Hypothesis samples: {len(hypothesis_wavelengths)}")
print(f"  Min: {min(hypothesis_wavelengths):.2f}, Max: {max(hypothesis_wavelengths):.2f}")
print(f"  Near boundaries (within 10nm): {sum(1 for v in hypothesis_wavelengths if v < 210 or v > 1090)}")
print(f"Naive random samples: {len(naive_wavelengths)}")
print(f"  Min: {min(naive_wavelengths):.2f}, Max: {max(naive_wavelengths):.2f}")
print(f"  Near boundaries (within 10nm): {sum(1 for v in naive_wavelengths if v < 210 or v > 1090)}")
print("-" * 50)

fig = make_subplots(rows=1, cols=2, subplot_titles=['Naive Random Sampling', 'Actual Hypothesis Sampling'])

fig.add_trace(go.Histogram(x=naive_wavelengths, nbinsx=30, name='Naive',
                           marker_color='#2BCDC1', opacity=0.7), row=1, col=1)
fig.add_trace(go.Histogram(x=hypothesis_wavelengths, nbinsx=30, name='Hypothesis',
                           marker_color='#F66095', opacity=0.7), row=1, col=2)

fig.update_layout(
    title='Value Distribution: Naive Random vs Actual Hypothesis Generation',
    template='plotly_dark',
    showlegend=False
)
fig.update_xaxes(title_text='Wavelength (nm)', row=1, col=1)
fig.update_xaxes(title_text='Wavelength (nm)', row=1, col=2)
fig.update_yaxes(title_text='Frequency', row=1, col=1)
fig.update_yaxes(title_text='Frequency', row=1, col=2)

from orgutils import plotly_figure_to_json, plotly_tight_layout
plotly_tight_layout(fig)
plotly_figure_to_json(fig, "../static/dm_hypothesis_distribution.json")

Sampling Comparison (Wavelength 200-1100nm):
--------------------------------------------------
Hypothesis samples: 500
  Min: 200.00, Max: 1100.00
  Near boundaries (within 10nm): 72
Naive random samples: 500
  Min: 204.56, Max: 1093.67
  Near boundaries (within 10nm): 8
--------------------------------------------------

def recalculate_curve(curve: CalibrationCurve, new_reading: CalibrationPoint) -> CalibrationCurve:
    """Add a new calibration point and recalculate the curve."""
    all_readings = curve.readings + [new_reading]
    slope, intercept, r_squared = fit_linear_regression(all_readings)

    # BUG: accidentally swapped slope and intercept
    return CalibrationCurve(
        readings=all_readings,
        r_squared=r_squared,
        slope=intercept,  # BUG: wrong assignment!
        intercept=slope   # BUG: wrong assignment!
    )

import icontract

@icontract.require(lambda curve: len(curve.readings) >= 2, "Need at least 2 points to fit a curve")
@icontract.require(lambda new_reading: new_reading.concentration >= 0, "Concentration must be non-negative")
def recalculate_curve(curve: CalibrationCurve, new_reading: CalibrationPoint) -> CalibrationCurve:
    """Add a new calibration point and recalculate the curve."""
    all_readings = curve.readings + [new_reading]
    slope, intercept, r_squared = fit_linear_regression(all_readings)

    return CalibrationCurve(
        readings=all_readings,
        r_squared=r_squared,
        slope=slope,
        intercept=intercept
    )

import icontract

@icontract.ensure(lambda result: 0 <= result.r_squared <= 1, "R² must be between 0 and 1")
@icontract.ensure(
    lambda curve, result: len(result.readings) == len(curve.readings) + 1,
    "Result must have exactly one more reading"
)
@icontract.ensure(
    lambda result: result.slope != 0 or all(r.response == result.intercept for r in result.readings),
    "Zero slope only valid if all responses equal intercept"
)
def recalculate_curve(curve: CalibrationCurve, new_reading: CalibrationPoint) -> CalibrationCurve:
    """Add a new calibration point and recalculate the curve."""
    all_readings = curve.readings + [new_reading]
    slope, intercept, r_squared = fit_linear_regression(all_readings)

    return CalibrationCurve(
        readings=all_readings,
        r_squared=r_squared,
        slope=slope,
        intercept=intercept
    )

import icontract
from pydantic import BaseModel
import numpy as np

def r_squared_matches_fit(self) -> bool:
    """Invariant: R² must be consistent with actual readings and coefficients."""
    if len(self.readings) < 2:
        return True  # Can't verify with insufficient data
    concentrations = np.array([r.concentration for r in self.readings])
    responses = np.array([r.response for r in self.readings])
    predicted = self.slope * concentrations + self.intercept
    ss_res = np.sum((responses - predicted) ** 2)
    ss_tot = np.sum((responses - np.mean(responses)) ** 2)
    calculated_r2 = 1 - (ss_res / ss_tot) if ss_tot > 0 else 1.0
    return abs(self.r_squared - calculated_r2) < 0.001  # Allow for floating point

@icontract.invariant(lambda self: len(self.readings) >= 2, "Calibration needs at least 2 points")
@icontract.invariant(r_squared_matches_fit, "R² must match actual fit quality")
class CalibrationCurve(BaseModel):
    readings: list[CalibrationPoint]
    r_squared: float
    slope: float
    intercept: float

    class Config:
        arbitrary_types_allowed = True

import icontract
from pydantic import BaseModel, field_validator
from typing import Optional
from enum import Enum
from datetime import datetime
import numpy as np

class QCStatus(str, Enum):
    PENDING = "pending"
    VALIDATED = "validated"
    FLAGGED = "flagged"
    REJECTED = "rejected"
    APPROVED = "approved"

class CalibrationPoint(BaseModel):
    concentration: float
    response: float
    replicate: int = 1

    @field_validator('concentration')
    @classmethod
    def validate_concentration(cls, v):
        if v < 0:
            raise ValueError('Concentration must be non-negative')
        return v

def r_squared_is_consistent(self) -> bool:
    """Invariant: R² must match the actual fit."""
    if len(self.readings) < 2:
        return True
    conc = np.array([r.concentration for r in self.readings])
    resp = np.array([r.response for r in self.readings])
    pred = self.slope * conc + self.intercept
    ss_res = np.sum((resp - pred) ** 2)
    ss_tot = np.sum((resp - np.mean(resp)) ** 2)
    calc_r2 = 1 - (ss_res / ss_tot) if ss_tot > 0 else 1.0
    return abs(self.r_squared - calc_r2) < 0.001

def approved_has_good_r_squared(self) -> bool:
    """Invariant: approved curves must have R² >= 0.99."""
    if self.status == QCStatus.APPROVED:
        return self.r_squared >= 0.99
    return True

@icontract.invariant(lambda self: len(self.readings) >= 2, "Need at least 2 calibration points")
@icontract.invariant(r_squared_is_consistent, "R² must match actual fit quality")
@icontract.invariant(approved_has_good_r_squared, "Approved curves need R² >= 0.99")
class CalibrationCurve(BaseModel):
    curve_id: str
    analyst_id: str
    readings: list[CalibrationPoint]
    slope: float
    intercept: float
    r_squared: float
    status: QCStatus = QCStatus.PENDING
    reviewer_notes: Optional[str] = None
    created_at: datetime

    class Config:
        arbitrary_types_allowed = True


# Function with contracts
@icontract.require(lambda curve: curve.status == QCStatus.PENDING,
                   "Can only validate pending curves")
@icontract.ensure(lambda result: result.status in [QCStatus.VALIDATED, QCStatus.FLAGGED],
                  "Validation must result in validated or flagged status")
def validate_curve(curve: CalibrationCurve) -> CalibrationCurve:
    """Validate a calibration curve based on R² threshold."""
    new_status = QCStatus.VALIDATED if curve.r_squared >= 0.99 else QCStatus.FLAGGED
    return CalibrationCurve(
        curve_id=curve.curve_id,
        analyst_id=curve.analyst_id,
        readings=curve.readings,
        slope=curve.slope,
        intercept=curve.intercept,
        r_squared=curve.r_squared,
        status=new_status,
        reviewer_notes=curve.reviewer_notes,
        created_at=curve.created_at
    )


@icontract.require(lambda curve: curve.status == QCStatus.VALIDATED,
                   "Can only approve validated curves")
@icontract.require(lambda reviewer_notes: reviewer_notes and reviewer_notes.strip(),
                   "Reviewer notes required for approval")
@icontract.ensure(lambda result: result.status == QCStatus.APPROVED,
                  "Curve must be approved after approval")
@icontract.ensure(lambda result: result.reviewer_notes is not None,
                  "Reviewer notes must be set")
def approve_curve(curve: CalibrationCurve, reviewer_notes: str) -> CalibrationCurve:
    """Approve a validated calibration curve."""
    return CalibrationCurve(
        curve_id=curve.curve_id,
        analyst_id=curve.analyst_id,
        readings=curve.readings,
        slope=curve.slope,
        intercept=curve.intercept,
        r_squared=curve.r_squared,
        status=QCStatus.APPROVED,
        reviewer_notes=reviewer_notes,
        created_at=curve.created_at
    )

"""
Integration tests demonstrating Polyfactory, Hypothesis, and icontract together.

This file is tangled from post-data-model-testing.org and can be run with:
    pytest test_data_model_integration.py -v
"""
from typing import Optional
from enum import Enum
from datetime import datetime

import numpy as np
import icontract
import pytest
from pydantic import BaseModel, field_validator
from polyfactory.factories.pydantic_factory import ModelFactory
from polyfactory import Use
from hypothesis import given, strategies as st, settings


# =============================================================================
# DOMAIN MODELS (with icontract invariants)
# =============================================================================

class QCStatus(str, Enum):
    PENDING = "pending"
    VALIDATED = "validated"
    FLAGGED = "flagged"
    REJECTED = "rejected"
    APPROVED = "approved"


class CalibrationPoint(BaseModel):
    concentration: float
    response: float
    replicate: int = 1

    @field_validator('concentration')
    @classmethod
    def validate_concentration(cls, v):
        if v < 0:
            raise ValueError('Concentration must be non-negative')
        return v


def r_squared_is_consistent(self) -> bool:
    """Invariant: R² must match the actual fit."""
    if len(self.readings) < 2:
        return True
    conc = np.array([r.concentration for r in self.readings])
    resp = np.array([r.response for r in self.readings])
    pred = self.slope * conc + self.intercept
    ss_res = np.sum((resp - pred) ** 2)
    ss_tot = np.sum((resp - np.mean(resp)) ** 2)
    calc_r2 = 1 - (ss_res / ss_tot) if ss_tot > 0 else 1.0
    return abs(self.r_squared - calc_r2) < 0.001


def approved_has_good_r_squared(self) -> bool:
    """Invariant: approved curves must have R² >= 0.99."""
    if self.status == QCStatus.APPROVED:
        return self.r_squared >= 0.99
    return True


@icontract.invariant(lambda self: len(self.readings) >= 2, "Need at least 2 calibration points")
@icontract.invariant(r_squared_is_consistent, "R² must match actual fit quality")
@icontract.invariant(approved_has_good_r_squared, "Approved curves need R² >= 0.99")
class CalibrationCurve(BaseModel):
    curve_id: str
    analyst_id: str
    readings: list[CalibrationPoint]
    slope: float
    intercept: float
    r_squared: float
    status: QCStatus = QCStatus.PENDING
    reviewer_notes: Optional[str] = None
    created_at: datetime

    class Config:
        arbitrary_types_allowed = True


# =============================================================================
# DOMAIN FUNCTIONS (with icontract pre/post conditions)
# =============================================================================

@icontract.require(lambda curve: curve.status == QCStatus.PENDING,
                   "Can only validate pending curves")
@icontract.ensure(lambda result: result.status in [QCStatus.VALIDATED, QCStatus.FLAGGED],
                  "Validation must result in validated or flagged status")
def validate_curve(curve: CalibrationCurve) -> CalibrationCurve:
    """Validate a calibration curve based on R² threshold."""
    new_status = QCStatus.VALIDATED if curve.r_squared >= 0.99 else QCStatus.FLAGGED
    return CalibrationCurve(
        curve_id=curve.curve_id,
        analyst_id=curve.analyst_id,
        readings=curve.readings,
        slope=curve.slope,
        intercept=curve.intercept,
        r_squared=curve.r_squared,
        status=new_status,
        reviewer_notes=curve.reviewer_notes,
        created_at=curve.created_at
    )


@icontract.require(lambda curve: curve.status == QCStatus.VALIDATED,
                   "Can only approve validated curves")
@icontract.require(lambda reviewer_notes: reviewer_notes and reviewer_notes.strip(),
                   "Reviewer notes required for approval")
@icontract.ensure(lambda result: result.status == QCStatus.APPROVED,
                  "Curve must be approved after approval")
def approve_curve(curve: CalibrationCurve, reviewer_notes: str) -> CalibrationCurve:
    """Approve a validated calibration curve."""
    return CalibrationCurve(
        curve_id=curve.curve_id,
        analyst_id=curve.analyst_id,
        readings=curve.readings,
        slope=curve.slope,
        intercept=curve.intercept,
        r_squared=curve.r_squared,
        status=QCStatus.APPROVED,
        reviewer_notes=reviewer_notes,
        created_at=curve.created_at
    )


# =============================================================================
# POLYFACTORY FACTORIES
# =============================================================================

class CalibrationPointFactory(ModelFactory):
    __model__ = CalibrationPoint
    # Constrain concentration to non-negative values (matching Pydantic validator)
    concentration = Use(lambda: abs(ModelFactory.__random__.uniform(0, 1000)))


class CalibrationCurveFactory(ModelFactory):
    __model__ = CalibrationCurve

    @classmethod
    def build(cls, **kwargs):
        # Generate readings that produce a valid fit
        readings = kwargs.get('readings') or [
            CalibrationPointFactory.build(concentration=float(i), response=float(i * 2.5 + 1.0))
            for i in range(5)
        ]
        # Calculate actual fit parameters
        conc = np.array([r.concentration for r in readings])
        resp = np.array([r.response for r in readings])
        slope, intercept = np.polyfit(conc, resp, 1)
        pred = slope * conc + intercept
        ss_res = np.sum((resp - pred) ** 2)
        ss_tot = np.sum((resp - np.mean(resp)) ** 2)
        r_squared = 1 - (ss_res / ss_tot) if ss_tot > 0 else 1.0

        return super().build(
            readings=readings,
            slope=slope,
            intercept=intercept,
            r_squared=r_squared,
            **{k: v for k, v in kwargs.items() if k not in ['readings', 'slope', 'intercept', 'r_squared']}
        )


# =============================================================================
# TESTS
# =============================================================================

class TestPolyfactoryCoverage:
    """Tests using Polyfactory's systematic coverage."""

    def test_qc_workflow_all_combinations(self):
        """Test QC workflow with polyfactory coverage - structural edge cases.

        Note: We iterate over QCStatus values manually because CalibrationCurve
        has complex invariants (R² consistency, minimum readings) that coverage()
        can't satisfy automatically. This demonstrates intentional structural
        coverage of the state machine.
        """
        tested_statuses = []

        for status in QCStatus:
            # Build a valid curve with this status
            curve = CalibrationCurveFactory.build(status=status)
            tested_statuses.append(status)

            if curve.status == QCStatus.PENDING:
                validated = validate_curve(curve)
                assert validated.status in [QCStatus.VALIDATED, QCStatus.FLAGGED]

                if validated.status == QCStatus.VALIDATED:
                    approved = approve_curve(validated, "Meets all QC criteria")
                    assert approved.status == QCStatus.APPROVED
                    assert approved.reviewer_notes is not None

        # Verify we tested all status values
        assert set(tested_statuses) == set(QCStatus)


class TestHypothesisProperties:
    """Property-based tests using Hypothesis."""

    @given(st.builds(
        CalibrationPoint,
        concentration=st.floats(min_value=0, max_value=1000, allow_nan=False),
        response=st.floats(min_value=0, max_value=10000, allow_nan=False),
        replicate=st.integers(min_value=1, max_value=10)
    ))
    @settings(max_examples=50)
    def test_calibration_point_concentration_non_negative(self, point: CalibrationPoint):
        """Hypothesis: concentration must be non-negative."""
        assert point.concentration >= 0

    @given(st.builds(
        CalibrationPoint,
        concentration=st.floats(min_value=0, max_value=1000, allow_nan=False),
        response=st.floats(min_value=0, max_value=10000, allow_nan=False),
        replicate=st.integers(min_value=1, max_value=10)
    ))
    @settings(max_examples=50)
    def test_calibration_point_response_is_finite(self, point: CalibrationPoint):
        """Hypothesis: response values are finite numbers."""
        assert np.isfinite(point.response)


class TestIcontractInvariants:
    """Tests verifying icontract catches invalid states."""

    def test_contracts_catch_invalid_r_squared(self):
        """Verify contracts catch scientifically invalid R² values."""
        readings = [
            CalibrationPoint(concentration=1.0, response=2.5),
            CalibrationPoint(concentration=2.0, response=5.0),
        ]

        # Try to create a curve with fake R² that doesn't match the data
        with pytest.raises(icontract.ViolationError) as exc_info:
            CalibrationCurve(
                curve_id="cal-001",
                analyst_id="analyst-1",
                readings=readings,
                slope=2.5,
                intercept=0.0,
                r_squared=0.5,  # Wrong! Actual R² is ~1.0
                created_at=datetime.now()
            )
        assert "R² must match actual fit quality" in str(exc_info.value)

    def test_contracts_require_minimum_readings(self):
        """Verify contracts require at least 2 calibration points."""
        with pytest.raises(icontract.ViolationError) as exc_info:
            CalibrationCurve(
                curve_id="cal-002",
                analyst_id="analyst-1",
                readings=[CalibrationPoint(concentration=1.0, response=2.5)],  # Only 1!
                slope=2.5,
                intercept=0.0,
                r_squared=1.0,
                created_at=datetime.now()
            )
        assert "Need at least 2 calibration points" in str(exc_info.value)

    def test_validate_requires_pending_status(self):
        """Verify validate_curve requires pending status."""
        readings = [
            CalibrationPointFactory.build(concentration=float(i), response=float(i * 2.5 + 1.0))
            for i in range(5)
        ]
        conc = np.array([r.concentration for r in readings])
        resp = np.array([r.response for r in readings])
        slope, intercept = np.polyfit(conc, resp, 1)

        curve = CalibrationCurve(
            curve_id="cal-003",
            analyst_id="analyst-1",
            readings=readings,
            slope=slope,
            intercept=intercept,
            r_squared=1.0,
            status=QCStatus.VALIDATED,  # Not pending!
            created_at=datetime.now()
        )

        with pytest.raises(icontract.ViolationError) as exc_info:
            validate_curve(curve)
        assert "Can only validate pending curves" in str(exc_info.value)

cd /Users/charlesbaker/svelte-projects/my-app/org && poetry run pytest test_data_model_integration.py -vvvv -q --disable-warnings --tb=short 2>&1

============================= test session starts ==============================
platform darwin -- Python 3.11.6, pytest-9.0.2, pluggy-1.6.0 -- /Users/charlesbaker/svelte-projects/my-app/org/.venv/bin/python
cachedir: .pytest_cache
hypothesis profile 'default'
rootdir: /Users/charlesbaker/svelte-projects/my-app/org
configfile: pyproject.toml
plugins: Faker-37.11.0, hypothesis-6.142.3
collecting ... collected 6 items

test_data_model_integration.py::TestPolyfactoryCoverage::test_qc_workflow_all_combinations PASSED [ 16%]
test_data_model_integration.py::TestHypothesisProperties::test_calibration_point_concentration_non_negative PASSED [ 33%]
test_data_model_integration.py::TestHypothesisProperties::test_calibration_point_response_is_finite PASSED [ 50%]
test_data_model_integration.py::TestIcontractInvariants::test_contracts_catch_invalid_r_squared PASSED [ 66%]
test_data_model_integration.py::TestIcontractInvariants::test_contracts_require_minimum_readings PASSED [ 83%]
test_data_model_integration.py::TestIcontractInvariants::test_validate_requires_pending_status PASSED [100%]

======================== 6 passed, 3 warnings in 1.40s =========================

import plotly.graph_objects as go

# =============================================================================
# SAFETY NET FUNNEL: Bug Survival Through Testing Layers
# =============================================================================
# This visualization models how bugs "survive" through successive testing layers.
#
# HYBRID APPROACH:
#   - If test_metrics was populated by running the "Combining Everything" tests,
#     we use empirical data to adjust the catch rates
#   - Otherwise, we fall back to research-based baseline estimates
#
# The base numbers come from our combinatorial analysis (7,680 combinations)
# =============================================================================

# Check if we have empirical data from running the tests
empirical_mode = False
try:
    if (test_metrics.get('polyfactory_combinations', 0) > 0 or
        test_metrics.get('hypothesis_examples', 0) > 0 or
        test_metrics.get('icontract_violations_caught', 0) > 0):
        empirical_mode = True
        print("Using EMPIRICAL data from test runs!")
except NameError:
    # test_metrics not defined - use baseline estimates
    test_metrics = {'polyfactory_combinations': 0, 'hypothesis_examples': 0, 'icontract_violations_caught': 0}

if not empirical_mode:
    print("Using BASELINE estimates (run 'Combining Everything' tests for empirical data)")

# Starting point: total structural state space from our earlier analysis
# (See "Growth Analysis" section: 2^4 × 3 × 4 × 5 × 2^3 = 7,680)
total_state_space = 7680

# Layer 1: Type Hints (static analysis)
# Research suggests type checkers find 15-40% of bugs depending on codebase
type_hint_catch_rate = 0.30
after_type_hints = int(total_state_space * (1 - type_hint_catch_rate))

# Layer 2: Pydantic Validation (runtime construction)
# Catches: type coercion failures, value range violations, missing required fields
pydantic_catch_rate = 0.45
after_pydantic = int(after_type_hints * (1 - pydantic_catch_rate))

# Layer 3: Polyfactory Coverage (structural edge cases)
# Empirical adjustment: more combinations tested = higher catch rate
polyfactory_combos = test_metrics.get('polyfactory_combinations', 0)
if empirical_mode and polyfactory_combos > 0:
    # Each combination tested catches ~5% of remaining structural bugs
    polyfactory_catch_rate = min(0.60, 0.15 + polyfactory_combos * 0.05)
else:
    polyfactory_catch_rate = 0.35  # Baseline: assumes ~3-5 combinations tested
after_polyfactory = int(after_pydantic * (1 - polyfactory_catch_rate))

# Layer 4: Hypothesis (value-level edge cases)
# Empirical adjustment: more examples = higher catch rate (diminishing returns)
hypothesis_examples = test_metrics.get('hypothesis_examples', 0)
if empirical_mode and hypothesis_examples > 0:
    # Logarithmic scaling: first examples are most valuable
    import math
    hypothesis_catch_rate = min(0.70, 0.20 + 0.15 * math.log10(hypothesis_examples + 1))
else:
    hypothesis_catch_rate = 0.40  # Baseline: assumes ~20 examples
after_hypothesis = int(after_polyfactory * (1 - hypothesis_catch_rate))

# Layer 5: icontract (domain invariants)
# Empirical adjustment: each caught violation represents a class of bugs prevented
icontract_catches = test_metrics.get('icontract_violations_caught', 0)
if empirical_mode and icontract_catches > 0:
    # Each violation caught suggests we're catching ~20% of invariant bugs
    icontract_catch_rate = min(0.80, 0.40 + icontract_catches * 0.20)
else:
    icontract_catch_rate = 0.60  # Baseline: assumes invariants catch most logic errors
after_icontract = max(1, int(after_hypothesis * (1 - icontract_catch_rate)))

# Sanity check: each layer should show progressively fewer bugs
assert after_type_hints < total_state_space
assert after_pydantic < after_type_hints
assert after_polyfactory < after_pydantic
assert after_hypothesis < after_polyfactory
assert after_icontract < after_hypothesis

fig = go.Figure()

# Create funnel - labels show empirical counts if available
if empirical_mode:
    layers = [
        f'Untested State Space ({total_state_space:,})',
        f'After Type Hints ({after_type_hints:,})',
        f'After Pydantic ({after_pydantic:,})',
        f'After Polyfactory [{polyfactory_combos} tested] ({after_polyfactory:,})',
        f'After Hypothesis [{hypothesis_examples} examples] ({after_hypothesis:,})',
        f'After icontract [{icontract_catches} caught] ({after_icontract:,})'
    ]
else:
    layers = [
        f'Untested State Space ({total_state_space:,})',
        f'After Type Hints ({after_type_hints:,})',
        f'After Pydantic ({after_pydantic:,})',
        f'After Polyfactory ({after_polyfactory:,})',
        f'After Hypothesis ({after_hypothesis:,})',
        f'After icontract ({after_icontract:,})'
    ]

values = [total_state_space, after_type_hints, after_pydantic,
          after_polyfactory, after_hypothesis, after_icontract]
colors = ['#ff6b6b', '#ffa502', '#ffd93d', '#6bcb77', '#4d96ff', '#9d65c9']

fig.add_trace(go.Funnel(
    y=layers,
    x=values,
    textposition="inside",
    textinfo="value+percent initial",
    marker=dict(color=colors),
    connector=dict(line=dict(color="royalblue", dash="dot", width=3))
))

fig.update_layout(
    title='Bug Survival Through Testing Layers',
    template='plotly_dark',
    font=dict(size=11),
    margin=dict(l=20, r=20, t=60, b=20)
)

# Print the calculation breakdown
mode_label = "EMPIRICAL" if empirical_mode else "BASELINE"
print(f"\nSafety Net Funnel Data ({mode_label}):")
print("-" * 60)
print(f"Total state space (from Growth Analysis):  {total_state_space:,}")
print(f"After Type Hints   ({type_hint_catch_rate*100:.0f}% catch):     {after_type_hints:,} ({100*after_type_hints/total_state_space:.0f}% remain)")
print(f"After Pydantic     ({pydantic_catch_rate*100:.0f}% catch):     {after_pydantic:,} ({100*after_pydantic/total_state_space:.0f}% remain)")
print(f"After Polyfactory  ({polyfactory_catch_rate*100:.0f}% catch):     {after_polyfactory:,} ({100*after_polyfactory/total_state_space:.0f}% remain)")
if empirical_mode:
    print(f"  └─ {polyfactory_combos} combinations tested")
print(f"After Hypothesis   ({hypothesis_catch_rate*100:.0f}% catch):     {after_hypothesis:,} ({100*after_hypothesis/total_state_space:.0f}% remain)")
if empirical_mode:
    print(f"  └─ {hypothesis_examples} examples generated")
print(f"After icontract    ({icontract_catch_rate*100:.0f}% catch):     {after_icontract:,} ({100*after_icontract/total_state_space:.0f}% remain)")
if empirical_mode:
    print(f"  └─ {icontract_catches} violations caught")
print("-" * 60)
print(f"Total bug reduction: {100*(1-after_icontract/total_state_space):.1f}%")

from orgutils import plotly_figure_to_json, plotly_tight_layout
plotly_tight_layout(fig)
plotly_figure_to_json(fig, "../static/dm_safety_funnel.json")
assert False  # Force output display

Using BASELINE estimates (run 'Combining Everything' tests for empirical data)

Safety Net Funnel Data (BASELINE):
------------------------------------------------------------
Total state space (from Growth Analysis):  7,680
After Type Hints   (30% catch):     5,376 (70% remain)
After Pydantic     (45% catch):     2,956 (38% remain)
After Polyfactory  (35% catch):     1,921 (25% remain)
After Hypothesis   (40% catch):     1,152 (15% remain)
After icontract    (60% catch):     460 (6% remain)
------------------------------------------------------------
Total bug reduction: 94.0%

flowchart TB
    subgraph DEV["Development Phase"]
        direction TB
        M1["Define Pydantic Models"]
        M2["Add icontract Invariants"]
        M3["Create Polyfactory Factories"]
        M1 --> M2 --> M3
    end

    subgraph TEST["Testing Phase"]
        direction TB
        T1["Unit Tests<br/>(Polyfactory coverage)"]
        T2["Property Tests<br/>(Hypothesis strategies)"]
        T3["Integration Tests<br/>(Contract verification)"]
        T1 --> T2 --> T3
    end

    subgraph CI["CI/CD Pipeline"]
        direction TB
        C1["Pre-commit Hooks<br/>mypy + icontract-lint"]
        C2["pytest + hypothesis<br/>--hypothesis-seed=CI"]
        C3["Coverage Reports<br/>structural + value"]
        C1 --> C2 --> C3
    end

    subgraph PROD["Production"]
        direction TB
        P1["Runtime Contract<br/>Checking (icontract)"]
        P2["Observability<br/>Contract Violations → Alerts"]
        P3["Data Quality<br/>Dashboards"]
        P1 --> P2 --> P3
    end

    DEV --> TEST --> CI --> PROD

    style DEV fill:#e1f5fe
    style TEST fill:#f3e5f5
    style CI fill:#e8f5e9
    style PROD fill:#fff3e0

"""Quick start template for systematic model testing."""
from datetime import datetime
from typing import Optional

import icontract
from hypothesis import given, strategies as st, settings
from pydantic import BaseModel
from polyfactory.factories.pydantic_factory import ModelFactory
from polyfactory import Use


# 1. Define your model with icontract invariants
def my_invariant(self) -> bool:
    """Document your domain rule here."""
    return True  # Replace with actual invariant logic


@icontract.invariant(my_invariant, "Description of invariant")
class MyModel(BaseModel):
    required_field: str
    optional_field: Optional[float] = None
    # Add your fields here


# 2. Create a factory with field constraints
class MyModelFactory(ModelFactory):
    __model__ = MyModel
    # Constrain fields that need specific ranges
    optional_field = Use(lambda: abs(ModelFactory.__random__.uniform(0, 100)))


# 3. Write structural coverage test
class TestMyModelCoverage:
    def test_all_structural_combinations(self):
        """Test all combinations of optional fields and enum values."""
        for instance in MyModelFactory.coverage():
            # Add assertions about valid instances
            assert instance.required_field  # Non-empty


# 4. Write property-based tests
class TestMyModelProperties:
    @given(st.builds(MyModel, required_field=st.text(min_size=1)))
    @settings(max_examples=100)
    def test_required_field_is_non_empty(self, instance: MyModel):
        """Property: required_field is never empty."""
        assert len(instance.required_field) > 0


# 5. Write contract verification tests
class TestMyModelContracts:
    def test_invariant_catches_invalid_state(self):
        """Verify invariants catch domain rule violations."""
        # Test that creating an invalid instance raises ViolationError
        pass  # Implement based on your invariant