first fuzz test

poetry.lock

@@ -1,4 +1,4 @@
-# This file is automatically @generated by Poetry 2.1.1 and should not be changed by hand.
+# This file is automatically @generated by Poetry 2.1.3 and should not be changed by hand.
 
 [[package]]
 name = "ago"
@@ -49,7 +49,7 @@ version = "25.3.0"
 description = "Classes Without Boilerplate"
 optional = false
 python-versions = ">=3.8"
-groups = ["dev"]
+groups = ["main", "dev"]
 files = [
     {file = "attrs-25.3.0-py3-none-any.whl", hash = "sha256:427318ce031701fea540783410126f03899a97ffc6f61596ad581ac2e40e3bc3"},
     {file = "attrs-25.3.0.tar.gz", hash = "sha256:75d7cefc7fb576747b2c81b4442d4d4a1ce0900973527c011d1030fd3bf4af1b"},
@@ -1433,6 +1433,40 @@ files = [
 [package.extras]
 tests = ["freezegun", "pytest", "pytest-cov"]
 
+[[package]]
+name = "hypothesis"
+version = "6.139.0"
+description = "A library for property-based testing"
+optional = false
+python-versions = ">=3.9"
+groups = ["main"]
+files = [
+    {file = "hypothesis-6.139.0-py3-none-any.whl", hash = "sha256:0943f8e910e79c9c368fbc758a2aa9db441a4802a96c3f51dd990070f5d1cb27"},
+    {file = "hypothesis-6.139.0.tar.gz", hash = "sha256:bc52c47dcc3150485674329b1fcfbfef17c3d38fdc84b02834624d9dcf4937a5"},
+]
+
+[package.dependencies]
+attrs = ">=22.2.0"
+sortedcontainers = ">=2.1.0,<3.0.0"
+
+[package.extras]
+all = ["black (>=20.8b0)", "click (>=7.0)", "crosshair-tool (>=0.0.95)", "django (>=4.2)", "dpcontracts (>=0.4)", "hypothesis-crosshair (>=0.0.25)", "lark (>=0.10.1)", "libcst (>=0.3.16)", "numpy (>=1.19.3)", "pandas (>=1.1)", "pytest (>=4.6)", "python-dateutil (>=1.4)", "pytz (>=2014.1)", "redis (>=3.0.0)", "rich (>=9.0.0)", "tzdata (>=2025.2) ; sys_platform == \"win32\" or sys_platform == \"emscripten\"", "watchdog (>=4.0.0)"]
+cli = ["black (>=20.8b0)", "click (>=7.0)", "rich (>=9.0.0)"]
+codemods = ["libcst (>=0.3.16)"]
+crosshair = ["crosshair-tool (>=0.0.95)", "hypothesis-crosshair (>=0.0.25)"]
+dateutil = ["python-dateutil (>=1.4)"]
+django = ["django (>=4.2)"]
+dpcontracts = ["dpcontracts (>=0.4)"]
+ghostwriter = ["black (>=20.8b0)"]
+lark = ["lark (>=0.10.1)"]
+numpy = ["numpy (>=1.19.3)"]
+pandas = ["pandas (>=1.1)"]
+pytest = ["pytest (>=4.6)"]
+pytz = ["pytz (>=2014.1)"]
+redis = ["redis (>=3.0.0)"]
+watchdog = ["watchdog (>=4.0.0)"]
+zoneinfo = ["tzdata (>=2025.2) ; sys_platform == \"win32\" or sys_platform == \"emscripten\""]
+
 [[package]]
 name = "identify"
 version = "2.6.13"
@@ -3710,7 +3744,7 @@ version = "2.4.0"
 description = "Sorted Containers -- Sorted List, Sorted Dict, Sorted Set"
 optional = false
 python-versions = "*"
-groups = ["dev"]
+groups = ["main", "dev"]
 files = [
     {file = "sortedcontainers-2.4.0-py2.py3-none-any.whl", hash = "sha256:a163dcaede0f1c021485e957a39245190e74249897e2ae4b2aa38595db237ee0"},
     {file = "sortedcontainers-2.4.0.tar.gz", hash = "sha256:25caa5a06cc30b6b83d11423433f65d1f9d76c4c6a0c90e3379eaa43b9bfdb88"},
@@ -4228,4 +4262,4 @@ cffi = ["cffi (>=1.17) ; python_version >= \"3.13\" and platform_python_implemen
 [metadata]
 lock-version = "2.1"
 python-versions = "^3.13.2"
-content-hash = "77a89315bb2b389ca93fd60a5974e144c946210e6418b7c2362dd6d235b95c2d"
+content-hash = "e00f86dc111537c6673d90b2f2a3be2d63b7ab4997447ef8e7538ff6a666e928"

pyproject.toml

@@ -20,6 +20,7 @@ flask-login = "^0.6"
 flask-talisman = "*"
 flask-wtf = "^1.2"
 gunicorn = {version = "==23.0.0", extras = ["gevent"]}
+hypothesis = "^6.138.15"
 humanize = "~=4.13"
 itsdangerous = "~=2.2"
 jinja2 = "^3.1.6"

tests/app/main/views/test_send.py

@@ -1,4 +1,6 @@
 # -*- coding: utf-8 -*-
+import csv
+import io
 import uuid
 from functools import partial
 from glob import glob
@@ -12,6 +14,8 @@ from zipfile import BadZipFile
 
 import pytest
 from flask import url_for
+from hypothesis import given
+from hypothesis import strategies as st
 from xlrd.biffh import XLRDError
 from xlrd.xldate import XLDateAmbiguous, XLDateError, XLDateNegative, XLDateTooLarge
 
@@ -102,6 +106,73 @@ test_spreadsheet_files = glob(path.join("tests", "spreadsheet_files", "*"))
 test_non_spreadsheet_files = glob(path.join("tests", "non_spreadsheet_files", "*"))
 
 
+valid_phone = st.text(
+    alphabet=st.characters(min_codepoint=ord("0"), max_codepoint=ord("9")),
+    min_size=10,
+    max_size=15,
+).map(
+    lambda s: s
+)  # simple digits only;
+
+invalid_phone = st.one_of(
+    st.text(
+        min_size=0,
+        max_size=9,
+        alphabet=st.characters(blacklist_characters="0123456789"),
+    ),
+    st.text(min_size=16, max_size=30),  # too long
+    st.text(alphabet=st.characters(max_codepoint=0x10FFFF), min_size=1, max_size=5),
+    st.text().filter(lambda s: any(c.isalpha() for c in s)),
+)
+
+phone_strategy = st.one_of(valid_phone, invalid_phone)
+
+message_strategy = st.text(
+    alphabet=st.characters(
+        blacklist_characters="",
+        min_codepoint=0x20,  # avoid control except newline
+        max_codepoint=0x10FFFF,
+    ),
+    min_size=0,
+    max_size=500,
+)
+rows_strategy = st.integers(min_value=0, max_value=50)
+
+
+@given(
+    rows=rows_strategy,
+    rows_data=st.lists(st.tuples(phone_strategy, message_strategy), max_size=50),
+)
+def test_fuzz_upload_csv_batch_sms_handles_bad_and_good_input(rows, rows_data, client):
+    output = io.StringIO()
+    writer = csv.writer(output)
+    writer.writerow(["phone_number", "message"])
+    for phone, msg in rows_data[:rows]:
+        writer.writerow([phone, msg])
+    csv_content = output.getvalue()
+
+    response = client.post(
+        url_for("send_batch_sms_upload"),
+        data={"csv_file": (io.BytesIO(csv_content.encode("utf-8")), "batch.csv")},
+        content_type="multipart/form-data",
+    )
+
+    assert (
+        response.status_code != 500
+    ), f"Server crash for CSV:\n{csv_content}\nResponse: {response.status_code}, {response.data}"
+
+    has_valid = any(
+        phone.isdigit() and 10 <= len(phone) <= 15 for phone, msg in rows_data[:rows]
+    )
+    if has_valid:
+        assert response.status_code in (200, 201, 400), "Unexpected status"
+        if response.status_code == 400:
+            assert b"errors" in response.data or b"invalid" in response.data.lower()
+    else:
+        assert response.status_code == 400
+        assert b"errors" in response.data or b"invalid" in response.data.lower()
+
+
 def test_show_correct_title_and_description_for_email_sender_type(
     client_request,
     fake_uuid,