#!/usr/bin/env python3
"""Phase 3: Form Logic Engine — test suite.
Tests:
- Server-side validation engine (3.2)
- Server-side conditional field engine (3.3)
- Calculated field engine (3.4)
- Multi-step form support (3.5)
- Client-side conditional rendering (3.6) - SDK JS verification
Run: python -m pytest tests/test_phase3.py -v
"""
import json
import os
import sys
import sqlite3
import tempfile
import unittest
import importlib
# Use shared test DB
import tests.test_shared_db
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import app.models
from app.app import app as flask_app
from app.services.form_logic import (
validate_submission,
evaluate_condition,
compute_field,
compute_all_fields,
resolve_multi_steps,
)
class TestEvaluateCondition(unittest.TestCase):
"""Test server-side conditional field engine (3.3)."""
def test_no_condition_always_visible(self):
field = {"key": "test"}
self.assertTrue(evaluate_condition(field, {}))
def test_eq_operator(self):
field = {"key": "spouse_name", "condition": {"field": "marital_status", "operator": "eq", "value": "married"}}
self.assertTrue(evaluate_condition(field, {"marital_status": "married"}))
self.assertFalse(evaluate_condition(field, {"marital_status": "single"}))
def test_neq_operator(self):
field = {"key": "other_detail", "condition": {"field": "choice", "operator": "neq", "value": "none"}}
self.assertTrue(evaluate_condition(field, {"choice": "something"}))
self.assertFalse(evaluate_condition(field, {"choice": "none"}))
def test_filled_operator(self):
field = {"key": "detail", "condition": {"field": "name", "operator": "filled"}}
self.assertTrue(evaluate_condition(field, {"name": "John"}))
self.assertFalse(evaluate_condition(field, {"name": ""}))
self.assertFalse(evaluate_condition(field, {}))
def test_empty_operator(self):
field = {"key": "help_text", "condition": {"field": "result", "operator": "empty"}}
self.assertTrue(evaluate_condition(field, {"result": ""}))
self.assertFalse(evaluate_condition(field, {"result": "done"}))
def test_numeric_gt(self):
field = {"key": "shipping", "condition": {"field": "total", "operator": "gt", "value": "100"}}
self.assertTrue(evaluate_condition(field, {"total": "150"}))
self.assertFalse(evaluate_condition(field, {"total": "50"}))
self.assertFalse(evaluate_condition(field, {"total": "100"}))
def test_numeric_gte(self):
field = {"key": "shipping", "condition": {"field": "total", "operator": "gte", "value": "100"}}
self.assertTrue(evaluate_condition(field, {"total": "100"}))
self.assertTrue(evaluate_condition(field, {"total": "150"}))
self.assertFalse(evaluate_condition(field, {"total": "99"}))
def test_numeric_lt(self):
field = {"key": "discount", "condition": {"field": "age", "operator": "lt", "value": "18"}}
self.assertTrue(evaluate_condition(field, {"age": "16"}))
self.assertFalse(evaluate_condition(field, {"age": "20"}))
def test_numeric_lte(self):
field = {"key": "discount", "condition": {"field": "age", "operator": "lte", "value": "18"}}
self.assertTrue(evaluate_condition(field, {"age": "18"}))
self.assertFalse(evaluate_condition(field, {"age": "19"}))
def test_contains(self):
field = {"key": "detail", "condition": {"field": "message", "operator": "contains", "value": "urgent"}}
self.assertTrue(evaluate_condition(field, {"message": "This is urgent"}))
self.assertFalse(evaluate_condition(field, {"message": "No rush today"}))
def test_not_contains(self):
field = {"key": "detail", "condition": {"field": "message", "operator": "not_contains", "value": "spam"}}
self.assertTrue(evaluate_condition(field, {"message": "Hello world"}))
self.assertFalse(evaluate_condition(field, {"message": "spam content"}))
def test_in_operator(self):
field = {"key": "warranty", "condition": {"field": "product", "operator": "in", "value": "laptop,phone,tablet"}}
self.assertTrue(evaluate_condition(field, {"product": "phone"}))
self.assertFalse(evaluate_condition(field, {"product": "shoes"}))
def test_not_in_operator(self):
field = {"key": "warranty", "condition": {"field": "product", "operator": "not_in", "value": "laptop,phone,tablet"}}
self.assertFalse(evaluate_condition(field, {"product": "phone"}))
self.assertTrue(evaluate_condition(field, {"product": "shoes"}))
def test_missing_target_value_returns_false(self):
field = {"key": "detail", "condition": {"field": "name", "operator": "filled"}}
self.assertFalse(evaluate_condition(field, {"other": "value"}))
def test_numeric_comparison_with_non_numeric_returns_false(self):
field = {"key": "x", "condition": {"field": "y", "operator": "gt", "value": "5"}}
self.assertFalse(evaluate_condition(field, {"y": "not_a_number"}))
class TestValidateSubmission(unittest.TestCase):
"""Test server-side validation engine (3.2)."""
def test_valid_email(self):
fields = [{"key": "email", "type": "email", "required": True, "validation": {"type": "email"}}]
data = {"email": "test@example.com"}
self.assertEqual(validate_submission(fields, data), [])
def test_invalid_email(self):
fields = [{"key": "email", "type": "email", "required": True, "validation": {"type": "email"}}]
data = {"email": "not-an-email"}
errors = validate_submission(fields, data)
self.assertTrue(len(errors) > 0)
self.assertIn("email", errors[0]["field"])
def test_required_field_missing(self):
fields = [{"key": "name", "type": "text", "required": True}]
data = {}
errors = validate_submission(fields, data)
self.assertTrue(len(errors) > 0)
def test_hidden_field_skips_validation(self):
fields = [
{"key": "service", "type": "select", "required": True},
{"key": "other_service", "type": "text", "required": True, "condition": {"field": "service", "operator": "eq", "value": "other"}},
]
# When service != 'other', other_service should be hidden and skip validation
data = {"service": "repair"}
errors = validate_submission(fields, data)
self.assertEqual(errors, [])
def test_hidden_field_validated_when_visible(self):
fields = [
{"key": "service", "type": "select", "required": True},
{"key": "other_service", "type": "text", "required": True, "condition": {"field": "service", "operator": "eq", "value": "other"}},
]
data = {"service": "other"}
errors = validate_submission(fields, data)
self.assertTrue(len(errors) > 0)
self.assertIn("other_service", errors[0]["field"])
def test_min_max_validation(self):
fields = [{"key": "qty", "type": "number", "min": 1, "max": 10}]
errors = validate_submission(fields, {"qty": "5"})
self.assertEqual(errors, [])
errors = validate_submission(fields, {"qty": "0"})
self.assertTrue(len(errors) > 0)
errors = validate_submission(fields, {"qty": "11"})
self.assertTrue(len(errors) > 0)
def test_pattern_validation(self):
fields = [{"key": "phone", "type": "phone", "validation": {"type": "pattern", "pattern": "^\\d{10}$"}}]
errors = validate_submission(fields, {"phone": "1234567890"})
self.assertEqual(errors, [])
errors = validate_submission(fields, {"phone": "abc"})
self.assertTrue(len(errors) > 0)
class TestComputeField(unittest.TestCase):
"""Test calculated field engine (3.4)."""
def test_basic_arithmetic(self):
field = {"key": "total", "calculation": {"formula": "{{price}} * {{quantity}}"}}
result = compute_field(field, {"price": "10", "quantity": "5"})
self.assertEqual(result, 50)
def test_rounding(self):
field = {"key": "avg", "calculation": {"formula": "({{a}} + {{b}}) / 2", "round": 1}}
result = compute_field(field, {"a": "10", "b": "11"})
self.assertEqual(result, 10.5)
def test_no_calculation(self):
field = {"key": "name"}
result = compute_field(field, {"name": "John"})
self.assertEqual(result, "John")
def test_missing_operand_returns_none(self):
field = {"key": "total", "calculation": {"formula": "{{x}} + {{y}}"}}
result = compute_field(field, {"x": "5"})
self.assertIsNone(result)
def test_compute_all_fields(self):
fields = [
{"key": "subtotal", "type": "computed", "calculation": {"formula": "{{qty}} * {{price}}"}},
{"key": "tax", "type": "computed", "calculation": {"formula": "{{subtotal}} * 0.1"}},
{"key": "total", "type": "computed", "calculation": {"formula": "{{subtotal}} + {{tax}}"}},
]
data = {"qty": "10", "price": "5"}
result = compute_all_fields(fields, data)
self.assertEqual(result["subtotal"], 50)
self.assertEqual(result["tax"], 5)
self.assertEqual(result["total"], 55)
class TestMultiStep(unittest.TestCase):
"""Test multi-step form support (3.5)."""
def test_resolve_steps(self):
fields = [
{"key": "name", "step": 1},
{"key": "email", "step": 1},
{"key": "phone", "step": 2},
{"key": "message", "step": 2},
]
result = resolve_multi_steps(fields)
self.assertEqual(len(result[1]), 2)
self.assertEqual(len(result[2]), 2)
def test_default_step_is_1(self):
fields = [
{"key": "name"},
{"key": "email", "step": 1},
]
result = resolve_multi_steps(fields)
self.assertEqual(len(result[1]), 2)
def test_empty_list_returns_empty(self):
result = resolve_multi_steps([])
self.assertEqual(result, {})
class TestConditionalRendering(unittest.TestCase):
"""Verify SDK conditional rendering code exists (3.6)."""
def test_sdk_has_evaluate_condition(self):
with open("app/static/embed.js") as f:
content = f.read()
self.assertIn("function evaluateCondition", content)
self.assertIn("function updateFieldVisibility", content)
self.assertIn("function wireConditionListeners", content)
self.assertIn("updateFieldVisibility(form, fieldConfig)", content)
def test_sdk_has_all_operators(self):
with open("app/static/embed.js") as f:
content = f.read()
for op in ["eq", "neq", "filled", "empty", "gt", "lt", "gte", "lte", "contains", "not_contains", "in", "not_in"]:
self.assertIn(f'\"{op}\"', content, f"Operator '{op}' missing from SDK condition evaluation")
def test_sdk_condition_wired_in_render(self):
with open("app/static/embed.js") as f:
content = f.read()
self.assertIn("wireConditionListeners(form, config)", content)
class TestRealTimeValidation(unittest.TestCase):
"""Verify SDK real-time validation (3.7)."""
def test_sdk_has_validate_field_function(self):
with open("app/static/embed.js") as f:
content = f.read()
self.assertIn("function validateField", content)
self.assertIn("function showFieldError", content)
self.assertIn("function clearFieldError", content)
self.assertIn("function wireRealTimeValidation", content)
def test_sdk_validation_wired_in_render(self):
with open("app/static/embed.js") as f:
content = f.read()
self.assertIn("wireRealTimeValidation(form, config, theme)", content)
def test_sdk_validation_listens_on_blur_and_input(self):
with open("app/static/embed.js") as f:
content = f.read()
self.assertIn("\"blur\"", content)
self.assertIn("\"input\"", content)
def test_sdk_handles_email_validation(self):
with open("app/static/embed.js") as f:
content = f.read()
self.assertIn("Please enter a valid email", content)
def test_sdk_handles_required_validation(self):
with open("app/static/embed.js") as f:
content = f.read()
self.assertIn("is required", content)
def test_sdk_handles_pattern_validation(self):
with open("app/static/embed.js") as f:
content = f.read()
self.assertIn("new RegExp", content)
def test_sdk_handles_min_max(self):
with open("app/static/embed.js") as f:
content = f.read()
self.assertIn("minMessage", content)
self.assertIn("maxMessage", content)
class TestMultiStepSdk(unittest.TestCase):
"""Verify SDK multi-step support (3.8)."""
def test_sdk_has_progress_bar(self):
with open("app/static/embed.js") as f:
content = f.read()
self.assertIn("af-progress", content)
self.assertIn("af-progress-bar", content)
self.assertIn("af-progress-segment", content)
def test_sdk_has_navigation(self):
with open("app/static/embed.js") as f:
content = f.read()
self.assertIn("af-prev", content)
self.assertIn("af-step-nav", content)
self.assertIn("nextBtn", content)
self.assertIn("prevBtn", content)
def test_sdk_step_management(self):
with open("app/static/embed.js") as f:
content = f.read()
self.assertIn("showStep", content)
self.assertIn("currentStep", content)
self.assertIn("updateProgress", content)
def test_sdk_step_css(self):
with open("app/static/embed.js") as f:
content = f.read()
self.assertIn(".af-step{", content)
self.assertIn(".af-step.active{", content)
def test_sdk_step_labels(self):
with open("app/static/embed.js") as f:
content = f.read()
self.assertIn("stepLabel", content)
self.assertIn("nextLabel", content)
self.assertIn("prevLabel", content)
class TestApiIntegration(unittest.TestCase):
"""Integration test: /api/submit with field validation."""
def setUp(self):
self.app = flask_app
self.app.config["TESTING"] = True
self.client = self.app.test_client()
with self.app.app_context():
# Create a test user first (foreign key requirement)
import hashlib
email_hash = hashlib.sha256(b"api_test@test.com").hexdigest()
conn = app.models.get_db()
conn.execute(
"INSERT OR IGNORE INTO users (email_hash, email_encrypted, password_hash, name_encrypted, tier) VALUES (?, ?, ?, ?, 'free')",
(email_hash, "api_test@test.com", "hashed", "API Test")
)
conn.commit()
# Find or create user
user = conn.execute("SELECT id FROM users WHERE email_hash = ?", (email_hash,)).fetchone()
user_id = user["id"] if user else 1
# Create a test site with field config
self.site = app.models.add_site("Test Validation", "test@test.com", user_id=user_id)
self.token = self.site["token"]
self.fields = [
{"key": "email", "type": "email", "required": True, "validation": {"type": "email"}},
{"key": "age", "type": "number", "min": 0, "max": 150},
]
app.models.update_site_fields(self.site["id"], self.fields)
def test_valid_submission(self):
resp = self.client.post(
f"/api/submit?token={self.token}",
data={"email": "valid@test.com", "age": "25"},
content_type="application/x-www-form-urlencoded",
)
self.assertEqual(resp.status_code, 200)
j = resp.get_json()
self.assertTrue(j.get("success"))
def test_invalid_email_returns_422(self):
resp = self.client.post(
f"/api/submit?token={self.token}",
data={"email": "not-an-email", "age": "25"},
content_type="application/x-www-form-urlencoded",
)
self.assertEqual(resp.status_code, 422)
def test_computed_fields_merged(self):
# Add a computed field
fields = [
{"key": "price", "type": "number"},
{"key": "qty", "type": "number"},
{"key": "total", "calculation": {"formula": "{{price}} * {{qty}}"}},
]
app.models.update_site_fields(self.site["id"], fields)
resp = self.client.post(
f"/api/submit?token={self.token}",
data={"price": "10", "qty": "5"},
content_type="application/x-www-form-urlencoded",
)
self.assertEqual(resp.status_code, 201)
class TestEditSiteUI(unittest.TestCase):
"""Verify edit_site.html has advanced sections (3.9)."""
def test_validation_section_exists(self):
with open("app/templates/edit_site.html") as f:
content = f.read()
self.assertIn('data-section="validation"', content)
def test_condition_section_exists(self):
with open("app/templates/edit_site.html") as f:
content = f.read()
self.assertIn('data-section="condition"', content)
def test_calculation_section_exists(self):
with open("app/templates/edit_site.html") as f:
content = f.read()
self.assertIn('data-section="calculation"', content)
def test_step_section_exists(self):
with open("app/templates/edit_site.html") as f:
content = f.read()
self.assertIn('data-section="step"', content)
def test_serialization_includes_conditions(self):
with open("app/templates/edit_site.html") as f:
content = f.read()
self.assertIn("data-cond-field", content)
self.assertIn("data-cond-operator", content)
self.assertIn("data-cond-value", content)
self.assertIn("data-calc-formula", content)
self.assertIn("data-calc-round", content)
class TestMigrationNormalize(unittest.TestCase):
"""Test field_config normalization migration (3.10)."""
def test_migration_adds_defaults(self):
conn = app.models.get_db()
# Insert a site with fields missing Phase 3 keys
site = conn.execute(
"INSERT INTO sites (token, name, owner_email, field_config) VALUES (?, ?, ?, ?)",
("mig-test-1", "Migration Test", "mig@test.com",
json.dumps([{"key": "name", "type": "text"}, {"key": "email", "type": "email"}])),
).lastrowid
conn.commit()
# Run migration
app.models.migrate_normalize_field_configs()
# Verify fields now have defaults
row = conn.execute("SELECT field_config FROM sites WHERE id = ?", (site,)).fetchone()
fields = json.loads(row["field_config"])
conn.close()
for field in fields:
self.assertIn("validation", field)
self.assertIsNone(field["validation"])
self.assertIn("condition", field)
self.assertIsNone(field["condition"])
self.assertIn("calculation", field)
self.assertIsNone(field["calculation"])
self.assertEqual(field["step"], 1)
self.assertEqual(field["required"], False)
def test_migration_is_idempotent(self):
conn = app.models.get_db()
site = conn.execute(
"INSERT INTO sites (token, name, owner_email, field_config) VALUES (?, ?, ?, ?)",
("mig-test-2", "Idempotent Test", "idm@test.com",
json.dumps([{"key": "x", "type": "text", "validation": None, "condition": None, "calculation": None, "step": 1, "required": False}])),
).lastrowid
conn.commit()
# Run migration twice
app.models.migrate_normalize_field_configs()
app.models.migrate_normalize_field_configs()
row = conn.execute("SELECT field_config FROM sites WHERE id = ?", (site,)).fetchone()
fields = json.loads(row["field_config"])
conn.close()
# Should still be the same
self.assertEqual(len(fields), 1)
self.assertEqual(fields[0]["key"], "x")
def test_migration_skips_none_config(self):
conn = app.models.get_db()
conn.execute(
"INSERT INTO sites (token, name, owner_email, field_config) VALUES (?, ?, ?, ?)",
("mig-test-3", "None Config", "none@test.com", None),
)
conn.commit()
# Should not crash
app.models.migrate_normalize_field_configs()
row = conn.execute("SELECT field_config FROM sites WHERE token = 'mig-test-3'").fetchone()
self.assertIsNone(row["field_config"])
conn.close()
if __name__ == "__main__":
unittest.main()