#!/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()