Fix SQLite concurrency for refresh rotation tests

0.4.27

@@ -0,0 +1 @@
+Requirement already satisfied: python-magic in c:\users\karamtot\agri vision\agri-vision\venv\lib\site-packages (0.4.27)

app.py

@@ -16,11 +16,11 @@
 from datetime import datetime
 from typing import Any, Dict, Optional, Tuple
 from werkzeug.utils import secure_filename
-from flask_limiter import Limiter
-from flask_limiter.util import get_remote_address
-from io import BytesIO
-from services.weather_service import get_weather
-from sqlalchemy import inspect, text
+from flask_limiter import Limiter
+from flask_limiter.util import get_remote_address
+from io import BytesIO
+from services.weather_service import get_weather
+from sqlalchemy import inspect, text
 
 import redis
 import base64
@@ -48,15 +48,15 @@
 from ultralytics import YOLO
 import json
 from jinja2 import Environment, FileSystemLoader
-from model_registry import registry
-from services.weather_service import generate_weather_recommendations
-from services.yield_service import estimate_yield
-from services.auth_security_service import (
-    AccountLockoutService,
-    get_client_ip,
-    get_user_agent,
-)
-from security_utils import (
+from model_registry import registry
+from services.weather_service import generate_weather_recommendations
+from services.yield_service import estimate_yield
+from services.auth_security_service import (
+    AccountLockoutService,
+    get_client_ip,
+    get_user_agent,
+)
+from security_utils import (
     UploadValidationError,
     cleanup_temp_upload,
     resolve_secret_key,
@@ -102,57 +102,90 @@
     storage_uri=limiter_storage_uri,
     strategy="fixed-window",
 )
-from models import db
-db.init_app(app)
-
-
-_account_lockout_schema_checked = False
-
-
-def ensure_account_lockout_schema() -> None:
-    """Backfill account lockout columns for existing create_all-managed DBs."""
-    inspector = inspect(db.engine)
-    if "users" not in inspector.get_table_names():
-        return
-
-    existing_columns = {column["name"] for column in inspector.get_columns("users")}
-    dialect = db.engine.dialect.name
-    datetime_type = "TIMESTAMP" if dialect == "postgresql" else "DATETIME"
-    columns = {
-        "failed_login_attempts": "INTEGER NOT NULL DEFAULT 0",
-        "last_failed_login_at": datetime_type,
-        "account_locked_until": datetime_type,
-        "last_successful_login_at": datetime_type,
-        "last_failed_ip": "VARCHAR(64)",
-        "last_successful_ip": "VARCHAR(64)",
-    }
-
-    changed = False
-    with db.engine.begin() as connection:
-        for column_name, ddl_type in columns.items():
-            if column_name not in existing_columns:
-                connection.execute(text(f"ALTER TABLE users ADD COLUMN {column_name} {ddl_type}"))
-                changed = True
-        connection.execute(
-            text(
-                "CREATE INDEX IF NOT EXISTS ix_users_account_locked_until "
-                "ON users (account_locked_until)"
-            )
-        )
-    if changed:
-        logger.info("Account lockout schema columns added to users table")
-
-
-@app.before_request
-def _ensure_account_lockout_schema_once() -> None:
-    global _account_lockout_schema_checked
-    if _account_lockout_schema_checked or app.config.get("TESTING"):
-        return
-    try:
-        ensure_account_lockout_schema()
-        _account_lockout_schema_checked = True
-    except Exception as exc:
-        logger.warning("Account lockout schema check skipped: %s", exc)
+from models import db
+
+# Load from the file next to this module so a stray ``sqlite_db`` on PYTHONPATH
+# cannot shadow the project helper (CI / odd environments).
+def _load_configure_sqlite_immediate_transactions():
+    import importlib.util
+    from pathlib import Path
+
+    path = Path(__file__).resolve().parent / "sqlite_db.py"
+    if not path.is_file():
+        raise ImportError(
+            f"sqlite_db.py is missing next to app.py ({path}). "
+            "Restore it from upstream; it defines configure_sqlite_immediate_transactions."
+        )
+    spec = importlib.util.spec_from_file_location("_agri_vision_sqlite_db", path)
+    if spec is None or spec.loader is None:
+        raise ImportError(f"Cannot load sqlite helpers from {path}")
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    fn = getattr(mod, "configure_sqlite_immediate_transactions", None)
+    if fn is None:
+        raise ImportError(
+            f"{path} does not define configure_sqlite_immediate_transactions"
+        )
+    return fn
+
+
+configure_sqlite_immediate_transactions = _load_configure_sqlite_immediate_transactions()
+
+db.init_app(app)
+
+
+_account_lockout_schema_checked = False
+
+
+def ensure_account_lockout_schema() -> None:
+    """Backfill account lockout columns for existing create_all-managed DBs."""
+    inspector = inspect(db.engine)
+    if "users" not in inspector.get_table_names():
+        return
+
+    existing_columns = {column["name"] for column in inspector.get_columns("users")}
+    dialect = db.engine.dialect.name
+    datetime_type = "TIMESTAMP" if dialect == "postgresql" else "DATETIME"
+    columns = {
+        "failed_login_attempts": "INTEGER NOT NULL DEFAULT 0",
+        "last_failed_login_at": datetime_type,
+        "account_locked_until": datetime_type,
+        "last_successful_login_at": datetime_type,
+        "last_failed_ip": "VARCHAR(64)",
+        "last_successful_ip": "VARCHAR(64)",
+    }
+
+    changed = False
+    with db.engine.begin() as connection:
+        for column_name, ddl_type in columns.items():
+            if column_name not in existing_columns:
+                connection.execute(text(f"ALTER TABLE users ADD COLUMN {column_name} {ddl_type}"))
+                changed = True
+        connection.execute(
+            text(
+                "CREATE INDEX IF NOT EXISTS ix_users_account_locked_until "
+                "ON users (account_locked_until)"
+            )
+        )
+    if changed:
+        logger.info("Account lockout schema columns added to users table")
+
+
+@app.before_request
+def _ensure_account_lockout_schema_once() -> None:
+    global _account_lockout_schema_checked
+    if _account_lockout_schema_checked or app.config.get("TESTING"):
+        return
+    try:
+        ensure_account_lockout_schema()
+        _account_lockout_schema_checked = True
+    except Exception as exc:
+        logger.warning("Account lockout schema check skipped: %s", exc)
+
+# Serialize concurrent writers on SQLite (e.g. refresh-token rotation).
+with app.app_context():
+    configure_sqlite_immediate_transactions(db.engine)
+
 
 # --- Login Manager Configuration ---
 login_manager = LoginManager()
@@ -597,14 +630,30 @@ def __call__(self, input_tensor: torch.Tensor, target_class_idx: Optional[int],
 ])
 
 
-def preprocess_image_for_resnet(image: np.ndarray) -> torch.Tensor:
+def preprocess_image_for_resnet(
+    image: np.ndarray,
+    target_size: Tuple[int, int] = (224, 224),
+) -> torch.Tensor:
     """Preprocess an RGB numpy image for ResNet50 inference.
 
-    Uses the module-level RESNET_TRANSFORM pipeline which includes
-    ImageNet normalization (mean=[0.485, 0.456, 0.406],
-    std=[0.229, 0.224, 0.225]).
+    Uses ImageNet normalization (mean=[0.485, 0.456, 0.406],
+    std=[0.229, 0.224, 0.225]). Default ``target_size`` matches the shared
+    ``RESNET_TRANSFORM`` pipeline; other sizes build an equivalent pipeline.
     """
-    return RESNET_TRANSFORM(image).unsqueeze(0)
+    if target_size == (224, 224):
+        return RESNET_TRANSFORM(image).unsqueeze(0)
+    transform = transforms.Compose(
+        [
+            transforms.ToPILImage(),
+            transforms.Resize(target_size),
+            transforms.ToTensor(),
+            transforms.Normalize(
+                mean=[0.485, 0.456, 0.406],
+                std=[0.229, 0.224, 0.225],
+            ),
+        ]
+    )
+    return transform(image).unsqueeze(0)
 
 
 def infer_disease(image):
@@ -2839,58 +2888,58 @@ def login():
     if current_user.is_authenticated:
         return redirect(url_for('index'))
 
-    if request.method == 'POST':
-        email = (request.form.get('email') or '').strip().lower()
-        password = request.form.get('password') or ''
-        remember = request.form.get('remember')
-        ip_address = get_client_ip()
-        user_agent = get_user_agent()
-        lockout_service = AccountLockoutService()
-
-        from models import User
-        user = User.query.filter_by(email=email).first()
-
-        if user:
-            lockout_state = lockout_service.check_lockout(user)
-            if lockout_state.unlocked_expired_lock:
-                lockout_service.record_unlock(
-                    user,
-                    ip=ip_address,
-                    user_agent=user_agent,
-                )
-                db.session.commit()
-            if lockout_state.locked:
-                flash('Account temporarily locked. Please try again later.', 'danger')
-                return render_template(
-                    'login.html',
-                    google_oauth_enabled=GOOGLE_OAUTH_ENABLED,
-                ), 423
-
-        if user and user.check_password(password):
-            if not user.is_active:
-                flash('Your account has been deactivated. Please contact support.', 'danger')
-                return render_template('login.html', google_oauth_enabled=GOOGLE_OAUTH_ENABLED)
-
-            login_user(user, remember=remember)
-            lockout_service.record_successful_login(
-                user,
-                ip=ip_address,
-                user_agent=user_agent,
-            )
-            user.last_login = user.last_successful_login_at
-            db.session.commit()
+    if request.method == 'POST':
+        email = (request.form.get('email') or '').strip().lower()
+        password = request.form.get('password') or ''
+        remember = request.form.get('remember')
+        ip_address = get_client_ip()
+        user_agent = get_user_agent()
+        lockout_service = AccountLockoutService()
+
+        from models import User
+        user = User.query.filter_by(email=email).first()
+
+        if user:
+            lockout_state = lockout_service.check_lockout(user)
+            if lockout_state.unlocked_expired_lock:
+                lockout_service.record_unlock(
+                    user,
+                    ip=ip_address,
+                    user_agent=user_agent,
+                )
+                db.session.commit()
+            if lockout_state.locked:
+                flash('Account temporarily locked. Please try again later.', 'danger')
+                return render_template(
+                    'login.html',
+                    google_oauth_enabled=GOOGLE_OAUTH_ENABLED,
+                ), 423
+
+        if user and user.check_password(password):
+            if not user.is_active:
+                flash('Your account has been deactivated. Please contact support.', 'danger')
+                return render_template('login.html', google_oauth_enabled=GOOGLE_OAUTH_ENABLED)
 
-            next_page = request.args.get('next')
-            return redirect(next_page) if next_page else redirect(url_for('index'))
-        else:
-            if user:
-                lockout_service.record_failed_login(
-                    user,
-                    ip=ip_address,
-                    user_agent=user_agent,
-                )
-                db.session.commit()
-            flash('Invalid email or password', 'danger')
+            login_user(user, remember=remember)
+            lockout_service.record_successful_login(
+                user,
+                ip=ip_address,
+                user_agent=user_agent,
+            )
+            user.last_login = user.last_successful_login_at
+            db.session.commit()
+
+            next_page = request.args.get('next')
+            return redirect(next_page) if next_page else redirect(url_for('index'))
+        else:
+            if user:
+                lockout_service.record_failed_login(
+                    user,
+                    ip=ip_address,
+                    user_agent=user_agent,
+                )
+                db.session.commit()
+            flash('Invalid email or password', 'danger')
 
     return render_template('login.html', google_oauth_enabled=GOOGLE_OAUTH_ENABLED)
 
@@ -3724,9 +3773,9 @@ def analyze_result():
 
     # Initialize database tables
     with app.app_context():
-        db.create_all()
-        ensure_account_lockout_schema()
-        logger.info("Database tables created")
+        db.create_all()
+        ensure_account_lockout_schema()
+        logger.info("Database tables created")
 
         # Seed enterprise RBAC (idempotent)
         try:

model_config.json

@@ -32,10 +32,10 @@
         "is_active": true,
         "ab_test_ratio": 0.0,
         "performance_metrics": {
-          "total_requests": 0,
-          "successful_predictions": 0,
-          "avg_confidence": 0.0,
-          "avg_inference_time": 0.0,
+          "total_requests": 6,
+          "successful_predictions": 6,
+          "avg_confidence": 0.9500000000000001,
+          "avg_inference_time": 0.00016689300537109375,
           "error_count": 0
         }
       }

models.py

@@ -105,6 +105,14 @@ class User(UserMixin, db.Model):
     last_failed_ip = db.Column(db.String(64), nullable=True)
     last_successful_ip = db.Column(db.String(64), nullable=True)
 
+    # Account lockout (see ensure_account_lockout_schema in app.py for legacy DB backfill)
+    failed_login_attempts = db.Column(db.Integer, default=0, nullable=False)
+    last_failed_login_at = db.Column(db.DateTime, nullable=True)
+    account_locked_until = db.Column(db.DateTime, nullable=True, index=True)
+    last_successful_login_at = db.Column(db.DateTime, nullable=True)
+    last_failed_ip = db.Column(db.String(64), nullable=True)
+    last_successful_ip = db.Column(db.String(64), nullable=True)
+
     # OAuth fields (populated when user signs in via Google)
     oauth_provider = db.Column(db.String(32), nullable=True)   # e.g. "google"
     oauth_id = db.Column(db.String(255), nullable=True, index=True)  # Provider's unique user ID

pr-body.md

@@ -0,0 +1,7 @@
+## Summary
+- Configure SQLite transactions with BEGIN IMMEDIATE so concurrent refresh rotation serializes like production row locking.
+- Harden refresh rotation tests: shared file DB, per-thread Flask app context, docstring and assertions.
+
+## Test plan
+- [ ] `pytest tests/test_refresh_rotation.py -v`
+- [ ] (Optional) `pytest tests/test_refresh_rotation.py::test_concurrent_refresh_only_one_succeeds --count=50 -v` if pytest-repeat is installed