From 63a0d89e10e47e1eba09b0739c4ba0a6f933b700 Mon Sep 17 00:00:00 2001
From: Zhao vistart <i@vistart.me>
Date: Sun, 8 Nov 2026 17:45:22 +0800
Subject: [PATCH 1/3] test(readonly): contract read-only and keyless models

Add sync/async cross-backend contracts for ReadOnlyMixin, the value-based write gate, every refused write path, reads that stay available, __primary_key__ = None, and UnaddressableRecordError. The contracts reuse the provider's users table and add no DDL.
---
 .../feature/basic/nopk/test_keyless_models.py | 187 ++++++++++++++++++
 .../basic/nopk/test_keyless_models_async.py   | 187 ++++++++++++++++++
 .../basic/readonly/test_readonly_models.py    | 169 ++++++++++++++++
 .../readonly/test_readonly_models_async.py    | 162 +++++++++++++++
 4 files changed, 705 insertions(+)
 create mode 100644 src/rhosocial/activerecord/testsuite/feature/basic/nopk/test_keyless_models.py
 create mode 100644 src/rhosocial/activerecord/testsuite/feature/basic/nopk/test_keyless_models_async.py
 create mode 100644 src/rhosocial/activerecord/testsuite/feature/basic/readonly/test_readonly_models.py
 create mode 100644 src/rhosocial/activerecord/testsuite/feature/basic/readonly/test_readonly_models_async.py

diff --git a/src/rhosocial/activerecord/testsuite/feature/basic/nopk/test_keyless_models.py b/src/rhosocial/activerecord/testsuite/feature/basic/nopk/test_keyless_models.py
new file mode 100644
index 0000000..41f2821
--- /dev/null
+++ b/src/rhosocial/activerecord/testsuite/feature/basic/nopk/test_keyless_models.py
@@ -0,0 +1,187 @@
+# src/rhosocial/activerecord/testsuite/feature/basic/nopk/test_keyless_models.py
+"""Keyless model contract (sync).
+
+``__primary_key__ = None`` is a declared shape: the relation has no single-row
+identity. Queries, ordering, grouping, aggregation and counting keep working;
+only identity-based access is unavailable, and it is refused with
+``UnaddressableRecordError`` instead of leaking a ``TypeError`` from
+``getattr``.
+
+This contract reuses the provider's existing ``users`` table, so it adds no
+DDL and stays backend-neutral. Keyless-ness is orthogonal to read-only-ness in
+both directions.
+"""
+
+from typing import ClassVar, Optional
+
+import pytest
+
+from rhosocial.activerecord.backend.errors import DatabaseError, UnaddressableRecordError
+from rhosocial.activerecord.base.field_proxy import FieldProxy
+from rhosocial.activerecord.field import ReadOnlyMixin
+from rhosocial.activerecord.model import ActiveRecord
+
+
+class KeylessUser(ActiveRecord):
+    """Mapped onto a keyless projection of the users table."""
+    __table_name__ = "users"
+    c: ClassVar[FieldProxy] = FieldProxy()
+    __primary_key__ = None
+    id: Optional[int] = None
+    username: str = ""
+
+
+class KeylessReadOnlyUser(ActiveRecord, ReadOnlyMixin):
+    __table_name__ = "users"
+    c: ClassVar[FieldProxy] = FieldProxy()
+    __primary_key__ = None
+    id: Optional[int] = None
+    username: str = ""
+
+
+class ReadOnlyWithKeyUser(ActiveRecord, ReadOnlyMixin):
+    __table_name__ = "users"
+    c: ClassVar[FieldProxy] = FieldProxy()
+    id: Optional[int] = None
+    username: str = ""
+
+
+class NormalUser(ActiveRecord):
+    __table_name__ = "users"
+    c: ClassVar[FieldProxy] = FieldProxy()
+    id: Optional[int] = None
+    username: str = ""
+
+
+@pytest.fixture
+def keyless_models(user_class):
+    backend = user_class.backend()
+    for model in (KeylessUser, KeylessReadOnlyUser, ReadOnlyWithKeyUser, NormalUser):
+        model.__backend__ = backend
+    yield user_class
+
+
+class TestKeylessDeclaration:
+    def test_none_is_an_accepted_declaration(self):
+        assert KeylessUser.primary_key() is None
+
+    def test_addressable_reports_false(self):
+        assert KeylessUser.addressable() is False
+
+    def test_addressable_reports_true_for_a_normal_model(self):
+        assert NormalUser.addressable() is True
+
+    def test_key_columns_are_empty_not_none(self):
+        assert KeylessUser.primary_key_columns() == ()
+        assert KeylessUser.primary_key_fields() == ()
+
+    def test_key_columns_never_contain_none(self):
+        for model in (KeylessUser, KeylessReadOnlyUser, ReadOnlyWithKeyUser, NormalUser):
+            assert None not in model.primary_key_columns(), model.__name__
+            assert None not in model.primary_key_fields(), model.__name__
+
+    def test_is_composite_pk_is_false(self):
+        assert KeylessUser.is_composite_pk() is False
+
+    def test_normal_model_is_unchanged(self):
+        assert NormalUser.primary_key_columns() == ("id",)
+        assert NormalUser.primary_key_fields() == ("id",)
+
+
+class TestKeylessIsNewRecord:
+    def test_a_constructed_record_is_new(self):
+        assert KeylessUser(username="x").is_new_record is True
+
+    def test_a_row_loaded_from_the_database_is_not_new(self):
+        loaded = KeylessUser(username="x")
+        loaded._is_from_db = True
+        assert loaded.is_new_record is False
+
+    def test_keyless_model_has_no_identity_to_inspect(self):
+        assert KeylessUser.primary_key_fields() == ()
+
+
+class TestKeylessQueriesKeepWorking:
+    def test_query_works(self, keyless_models):
+        assert isinstance(KeylessUser.query().all(), list)
+
+    def test_count_works(self, keyless_models):
+        assert KeylessUser.query().count() >= 0
+
+    def test_where_works(self, keyless_models):
+        assert isinstance(KeylessUser.query().where(KeylessUser.c.id == 1).all(), list)
+
+    def test_order_by_works(self, keyless_models):
+        assert isinstance(KeylessUser.query().order_by(KeylessUser.c.id).all(), list)
+
+
+class TestKeylessIdentityAccessIsRefused:
+    def test_find_one_by_value(self, keyless_models):
+        with pytest.raises(UnaddressableRecordError, match="find_one"):
+            KeylessUser.find_one(1)
+
+    def test_find_all_by_value_list(self, keyless_models):
+        with pytest.raises(UnaddressableRecordError, match="find_all"):
+            KeylessUser.find_all([1, 2])
+
+    def test_refresh(self):
+        with pytest.raises(UnaddressableRecordError, match="refresh"):
+            KeylessUser(username="x").refresh()
+
+    def test_the_error_names_model_and_operation(self, keyless_models):
+        with pytest.raises(UnaddressableRecordError) as excinfo:
+            KeylessUser.find_one(1)
+        assert excinfo.value.model_name == "KeylessUser"
+        assert excinfo.value.operation == "find_one"
+
+    def test_the_error_is_a_database_error(self):
+        assert issubclass(UnaddressableRecordError, DatabaseError)
+
+    def test_the_error_suggests_the_alternative(self, keyless_models):
+        with pytest.raises(UnaddressableRecordError) as excinfo:
+            KeylessUser.find_one(1)
+        assert "where(...)" in str(excinfo.value)
+
+
+class TestKeylessBulkIdentityIsRefused:
+    def test_bulk_update_refused(self, keyless_models):
+        with pytest.raises(UnaddressableRecordError, match="bulk_update"):
+            KeylessUser.bulk_update([KeylessUser(username="x")], ["username"])
+
+    def test_bulk_delete_refused(self, keyless_models):
+        with pytest.raises(UnaddressableRecordError, match="bulk_delete"):
+            KeylessUser.bulk_delete([KeylessUser(username="x")])
+
+
+class TestKeylessNonIdentityConditionsStillWork:
+    def test_find_all_with_no_condition(self, keyless_models):
+        assert isinstance(KeylessUser.find_all(), list)
+
+    def test_find_one_with_a_predicate(self, keyless_models):
+        # A predicate is not a primary-key condition, so it is not refused;
+        # the query simply runs. The users table may be empty, which is why
+        # only the absence of the refusal is asserted.
+        result = KeylessUser.find_one(KeylessUser.c.id == 1)
+        assert result is None or result.id == 1
+
+
+class TestKeylessOrthogonality:
+    def test_keyless_and_read_only_together(self, keyless_models):
+        assert KeylessReadOnlyUser.addressable() is False
+        assert KeylessReadOnlyUser.read_only() is True
+
+    def test_keyless_read_only_refuses_writes(self, keyless_models):
+        from rhosocial.activerecord.backend.errors import ReadOnlyError
+
+        with pytest.raises(ReadOnlyError):
+            KeylessReadOnlyUser(username="x").save()
+
+    def test_read_only_but_addressable(self, keyless_models):
+        assert ReadOnlyWithKeyUser.addressable() is True
+        assert ReadOnlyWithKeyUser.read_only() is True
+
+    def test_read_only_with_a_key_still_refuses_by_identity(self, keyless_models):
+        from rhosocial.activerecord.backend.errors import ReadOnlyError
+
+        with pytest.raises(ReadOnlyError):
+            ReadOnlyWithKeyUser(username="x").save()
diff --git a/src/rhosocial/activerecord/testsuite/feature/basic/nopk/test_keyless_models_async.py b/src/rhosocial/activerecord/testsuite/feature/basic/nopk/test_keyless_models_async.py
new file mode 100644
index 0000000..3837df3
--- /dev/null
+++ b/src/rhosocial/activerecord/testsuite/feature/basic/nopk/test_keyless_models_async.py
@@ -0,0 +1,187 @@
+# src/rhosocial/activerecord/testsuite/feature/basic/nopk/test_keyless_models_async.py
+"""Keyless model contract (async).
+
+``__primary_key__ = None`` is a declaration, not an async concern. The same
+decision points answer for the async family: reads keep working, identity-based
+access is refused, and keyless-ness stays orthogonal to read-only-ness.
+"""
+
+from typing import ClassVar, Optional
+
+import pytest
+
+from rhosocial.activerecord.backend.errors import (
+    DatabaseError,
+    ReadOnlyError,
+    UnaddressableRecordError,
+)
+from rhosocial.activerecord.base.field_proxy import FieldProxy
+from rhosocial.activerecord.field import ReadOnlyMixin
+from rhosocial.activerecord.model import AsyncActiveRecord
+
+
+class AsyncKeylessUser(AsyncActiveRecord):
+    __table_name__ = "users"
+    c: ClassVar[FieldProxy] = FieldProxy()
+    __primary_key__ = None
+    id: Optional[int] = None
+    username: str = ""
+
+
+class AsyncKeylessReadOnlyUser(AsyncActiveRecord, ReadOnlyMixin):
+    __table_name__ = "users"
+    c: ClassVar[FieldProxy] = FieldProxy()
+    __primary_key__ = None
+    id: Optional[int] = None
+    username: str = ""
+
+
+class AsyncReadOnlyWithKeyUser(AsyncActiveRecord, ReadOnlyMixin):
+    __table_name__ = "users"
+    c: ClassVar[FieldProxy] = FieldProxy()
+    id: Optional[int] = None
+    username: str = ""
+
+
+class AsyncNormalUser(AsyncActiveRecord):
+    __table_name__ = "users"
+    c: ClassVar[FieldProxy] = FieldProxy()
+    id: Optional[int] = None
+    username: str = ""
+
+
+@pytest.fixture
+async def async_keyless_models(async_user_class):
+    backend = async_user_class.backend()
+    for model in (
+        AsyncKeylessUser,
+        AsyncKeylessReadOnlyUser,
+        AsyncReadOnlyWithKeyUser,
+        AsyncNormalUser,
+    ):
+        model.__backend__ = backend
+    yield async_user_class
+
+
+class TestAsyncKeylessDeclaration:
+    def test_none_is_an_accepted_declaration(self):
+        assert AsyncKeylessUser.primary_key() is None
+
+    def test_addressable_reports_false(self):
+        assert AsyncKeylessUser.addressable() is False
+
+    def test_addressable_reports_true_for_a_normal_model(self):
+        assert AsyncNormalUser.addressable() is True
+
+    def test_key_columns_are_empty_not_none(self):
+        assert AsyncKeylessUser.primary_key_columns() == ()
+        assert AsyncKeylessUser.primary_key_fields() == ()
+
+    def test_key_columns_never_contain_none(self):
+        for model in (
+            AsyncKeylessUser,
+            AsyncKeylessReadOnlyUser,
+            AsyncReadOnlyWithKeyUser,
+            AsyncNormalUser,
+        ):
+            assert None not in model.primary_key_columns(), model.__name__
+            assert None not in model.primary_key_fields(), model.__name__
+
+    def test_is_composite_pk_is_false(self):
+        assert AsyncKeylessUser.is_composite_pk() is False
+
+    def test_normal_model_is_unchanged(self):
+        assert AsyncNormalUser.primary_key_columns() == ("id",)
+        assert AsyncNormalUser.primary_key_fields() == ("id",)
+
+
+class TestAsyncKeylessIsNewRecord:
+    def test_a_constructed_record_is_new(self):
+        assert AsyncKeylessUser(username="x").is_new_record is True
+
+    def test_a_row_loaded_from_the_database_is_not_new(self):
+        loaded = AsyncKeylessUser(username="x")
+        loaded._is_from_db = True
+        assert loaded.is_new_record is False
+
+    def test_keyless_model_has_no_identity_to_inspect(self):
+        assert AsyncKeylessUser.primary_key_fields() == ()
+
+
+class TestAsyncKeylessQueriesKeepWorking:
+    async def test_query_works(self, async_keyless_models):
+        assert isinstance(await AsyncKeylessUser.query().all(), list)
+
+    async def test_count_works(self, async_keyless_models):
+        assert await AsyncKeylessUser.query().count() >= 0
+
+    async def test_where_works(self, async_keyless_models):
+        assert isinstance(await AsyncKeylessUser.query().where(AsyncKeylessUser.c.id == 1).all(), list)
+
+    async def test_order_by_works(self, async_keyless_models):
+        assert isinstance(await AsyncKeylessUser.query().order_by(AsyncKeylessUser.c.id).all(), list)
+
+
+class TestAsyncKeylessIdentityAccessIsRefused:
+    async def test_find_one_by_value(self, async_keyless_models):
+        with pytest.raises(UnaddressableRecordError, match="find_one"):
+            await AsyncKeylessUser.find_one(1)
+
+    async def test_find_all_by_value_list(self, async_keyless_models):
+        with pytest.raises(UnaddressableRecordError, match="find_all"):
+            await AsyncKeylessUser.find_all([1, 2])
+
+    async def test_refresh(self):
+        with pytest.raises(UnaddressableRecordError, match="refresh"):
+            await AsyncKeylessUser(username="x").refresh()
+
+    async def test_the_error_names_model_and_operation(self, async_keyless_models):
+        with pytest.raises(UnaddressableRecordError) as excinfo:
+            await AsyncKeylessUser.find_one(1)
+        assert excinfo.value.model_name == "AsyncKeylessUser"
+        assert excinfo.value.operation == "find_one"
+
+    def test_the_error_is_a_database_error(self):
+        assert issubclass(UnaddressableRecordError, DatabaseError)
+
+    async def test_the_error_suggests_the_alternative(self, async_keyless_models):
+        with pytest.raises(UnaddressableRecordError) as excinfo:
+            await AsyncKeylessUser.find_one(1)
+        assert "where(...)" in str(excinfo.value)
+
+
+class TestAsyncKeylessBulkIdentityIsRefused:
+    async def test_bulk_update_refused(self, async_keyless_models):
+        with pytest.raises(UnaddressableRecordError, match="bulk_update"):
+            await AsyncKeylessUser.bulk_update([AsyncKeylessUser(username="x")], ["username"])
+
+    async def test_bulk_delete_refused(self, async_keyless_models):
+        with pytest.raises(UnaddressableRecordError, match="bulk_delete"):
+            await AsyncKeylessUser.bulk_delete([AsyncKeylessUser(username="x")])
+
+
+class TestAsyncKeylessNonIdentityConditionsStillWork:
+    async def test_find_all_with_no_condition(self, async_keyless_models):
+        assert isinstance(await AsyncKeylessUser.find_all(), list)
+
+    async def test_find_one_with_a_predicate(self, async_keyless_models):
+        result = await AsyncKeylessUser.find_one(AsyncKeylessUser.c.id == 1)
+        assert result is None or result.id == 1
+
+
+class TestAsyncKeylessOrthogonality:
+    async def test_keyless_and_read_only_together(self, async_keyless_models):
+        assert AsyncKeylessReadOnlyUser.addressable() is False
+        assert AsyncKeylessReadOnlyUser.read_only() is True
+
+    async def test_keyless_read_only_refuses_writes(self, async_keyless_models):
+        with pytest.raises(ReadOnlyError):
+            await AsyncKeylessReadOnlyUser(username="x").save()
+
+    async def test_read_only_but_addressable(self, async_keyless_models):
+        assert AsyncReadOnlyWithKeyUser.addressable() is True
+        assert AsyncReadOnlyWithKeyUser.read_only() is True
+
+    async def test_read_only_with_a_key_still_refuses_by_identity(self, async_keyless_models):
+        with pytest.raises(ReadOnlyError):
+            await AsyncReadOnlyWithKeyUser(username="x").save()
diff --git a/src/rhosocial/activerecord/testsuite/feature/basic/readonly/test_readonly_models.py b/src/rhosocial/activerecord/testsuite/feature/basic/readonly/test_readonly_models.py
new file mode 100644
index 0000000..4752e6d
--- /dev/null
+++ b/src/rhosocial/activerecord/testsuite/feature/basic/readonly/test_readonly_models.py
@@ -0,0 +1,169 @@
+# src/rhosocial/activerecord/testsuite/feature/basic/readonly/test_readonly_models.py
+"""Read-only model contract (sync).
+
+``ReadOnlyMixin`` is opt-in and orthogonal: read-only-ness is a static
+declaration independent of the primary key and of the physical object the model
+maps. A read-only table may have an auto-generated key, and a keyless table may
+be writable.
+
+This contract deliberately reuses the provider's existing ``users`` table and
+creates nothing. That keeps it backend-neutral: the assertions are about
+metadata, the gate ordering, and the error type, all of which are decided
+before SQL is issued. The "no SQL reached the database" assertion is
+SQLite-specific and lives in the core suite, not here.
+"""
+
+from typing import ClassVar, Optional
+
+import pytest
+
+from rhosocial.activerecord.backend.errors import DatabaseError, ReadOnlyError
+from rhosocial.activerecord.base import ObjectDeclaration
+from rhosocial.activerecord.base.field_proxy import FieldProxy
+from rhosocial.activerecord.field import ReadOnlyMixin
+from rhosocial.activerecord.interface import IReadOnlyBehavior
+from rhosocial.activerecord.model import ActiveRecord
+
+
+class ReadOnlyUser(ActiveRecord, ReadOnlyMixin):
+    __table_name__ = "users"
+    c: ClassVar[FieldProxy] = FieldProxy()
+    id: Optional[int] = None
+    username: str = ""
+
+
+class WritableUser(ActiveRecord):
+    __table_name__ = "users"
+    c: ClassVar[FieldProxy] = FieldProxy()
+    id: Optional[int] = None
+    username: str = ""
+
+
+class OptedOutUser(ActiveRecord, ReadOnlyMixin):
+    __table_name__ = "users"
+    c: ClassVar[FieldProxy] = FieldProxy()
+    id: Optional[int] = None
+    username: str = ""
+    __read_only__ = False
+
+
+class ImplementsButWritable(ActiveRecord, IReadOnlyBehavior):
+    __table_name__ = "users"
+    c: ClassVar[FieldProxy] = FieldProxy()
+    id: Optional[int] = None
+    username: str = ""
+
+
+class ReadOnlyView(ActiveRecord, ReadOnlyMixin):
+    """A read-only model that declares it maps a view."""
+
+    __table_name__ = "users"
+    c: ClassVar[FieldProxy] = FieldProxy()
+    id: Optional[int] = None
+    username: str = ""
+
+    @classmethod
+    def object_type(cls) -> ObjectDeclaration:
+        return ObjectDeclaration.VIEW
+
+
+@pytest.fixture
+def readonly_models(user_class):
+    backend = user_class.backend()
+    for model in (
+        ReadOnlyUser,
+        WritableUser,
+        OptedOutUser,
+        ImplementsButWritable,
+        ReadOnlyView,
+    ):
+        model.__backend__ = backend
+    yield user_class
+
+
+class TestReadonlyDeclaration:
+    """The declaration, not a type test, decides."""
+
+    def test_mixin_satisfies_the_interface(self):
+        assert issubclass(ReadOnlyMixin, IReadOnlyBehavior)
+        assert issubclass(ReadOnlyUser, IReadOnlyBehavior)
+
+    def test_read_only_reports_true(self):
+        assert ReadOnlyUser.read_only() is True
+
+    def test_writable_model_has_no_read_only_at_all(self):
+        assert not hasattr(WritableUser, "read_only")
+
+    def test_writable_model_is_never_refused(self):
+        WritableUser.refuse_read_only("save")
+
+    def test_explicit_false_opts_back_out(self):
+        assert OptedOutUser.read_only() is False
+        OptedOutUser.refuse_read_only("save")
+
+    def test_interface_implementor_reporting_false_still_writes(self):
+        assert issubclass(ImplementsButWritable, IReadOnlyBehavior)
+        assert ImplementsButWritable.read_only() is False
+        ImplementsButWritable.refuse_read_only("save")
+
+    def test_read_only_is_orthogonal_to_the_primary_key(self):
+        assert ReadOnlyUser.primary_key_columns() == ("id",)
+        assert WritableUser.primary_key_columns() == ("id",)
+
+    def test_read_only_model_can_declare_a_view_object_type(self):
+        assert ReadOnlyUser.object_type() is ObjectDeclaration.TABLE
+        assert ReadOnlyView.object_type() is ObjectDeclaration.VIEW
+
+
+class TestReadonlyWriteRefusal:
+    def test_save_raises(self, readonly_models):
+        with pytest.raises(ReadOnlyError, match="save is not allowed on read-only model"):
+            ReadOnlyUser(username="x").save()
+
+    def test_delete_raises(self, readonly_models):
+        record = ReadOnlyUser(id=1, username="x")
+        record._is_from_db = True
+        with pytest.raises(ReadOnlyError, match="delete is not allowed"):
+            record.delete()
+
+    def test_error_names_model_and_operation(self):
+        with pytest.raises(ReadOnlyError) as excinfo:
+            ReadOnlyUser.refuse_read_only("bulk_delete")
+        assert excinfo.value.model_name == "ReadOnlyUser"
+        assert excinfo.value.operation == "bulk_delete"
+
+    def test_error_is_a_database_error(self):
+        assert issubclass(ReadOnlyError, DatabaseError)
+
+
+class TestReadonlyClassLevelRefusal:
+    def test_bulk_create_raises(self, readonly_models):
+        with pytest.raises(ReadOnlyError, match="bulk_create"):
+            ReadOnlyUser.bulk_create([ReadOnlyUser(username="x")])
+
+    def test_bulk_update_raises(self, readonly_models):
+        with pytest.raises(ReadOnlyError, match="bulk_update"):
+            ReadOnlyUser.bulk_update([ReadOnlyUser(id=1, username="x")], ["username"])
+
+    def test_bulk_delete_raises(self, readonly_models):
+        with pytest.raises(ReadOnlyError, match="bulk_delete"):
+            ReadOnlyUser.bulk_delete([ReadOnlyUser(id=1, username="x")])
+
+    def test_update_all_raises(self, readonly_models):
+        with pytest.raises(ReadOnlyError, match="update_all"):
+            ReadOnlyUser.query().where(ReadOnlyUser.c.id == 1).update_all({"username": "y"})
+
+    def test_delete_all_raises(self, readonly_models):
+        with pytest.raises(ReadOnlyError, match="delete_all"):
+            ReadOnlyUser.query().where(ReadOnlyUser.c.id == 1).delete_all()
+
+
+class TestReadonlyReadsStillWork:
+    def test_query_works(self, readonly_models):
+        assert isinstance(ReadOnlyUser.query().all(), list)
+
+    def test_count_works(self, readonly_models):
+        assert ReadOnlyUser.query().count() >= 0
+
+    def test_where_works(self, readonly_models):
+        assert isinstance(ReadOnlyUser.query().where(ReadOnlyUser.c.id == 1).all(), list)
diff --git a/src/rhosocial/activerecord/testsuite/feature/basic/readonly/test_readonly_models_async.py b/src/rhosocial/activerecord/testsuite/feature/basic/readonly/test_readonly_models_async.py
new file mode 100644
index 0000000..38bf0a1
--- /dev/null
+++ b/src/rhosocial/activerecord/testsuite/feature/basic/readonly/test_readonly_models_async.py
@@ -0,0 +1,162 @@
+# src/rhosocial/activerecord/testsuite/feature/basic/readonly/test_readonly_models_async.py
+"""Read-only model contract (async).
+
+``ReadOnlyMixin`` has no async variant because ``read_only()`` is a zero-I/O
+predicate. This mirror proves the same single mixin gates every async write
+path and that the declaration surface is identical.
+"""
+
+from typing import ClassVar, Optional
+
+import pytest
+
+from rhosocial.activerecord.backend.errors import DatabaseError, ReadOnlyError
+from rhosocial.activerecord.base import ObjectDeclaration
+from rhosocial.activerecord.base.field_proxy import FieldProxy
+from rhosocial.activerecord.field import ReadOnlyMixin
+from rhosocial.activerecord.interface import IReadOnlyBehavior
+from rhosocial.activerecord.model import AsyncActiveRecord
+
+
+class AsyncReadOnlyUser(AsyncActiveRecord, ReadOnlyMixin):
+    __table_name__ = "users"
+    c: ClassVar[FieldProxy] = FieldProxy()
+    id: Optional[int] = None
+    username: str = ""
+
+
+class AsyncWritableUser(AsyncActiveRecord):
+    __table_name__ = "users"
+    c: ClassVar[FieldProxy] = FieldProxy()
+    id: Optional[int] = None
+    username: str = ""
+
+
+class AsyncOptedOutUser(AsyncActiveRecord, ReadOnlyMixin):
+    __table_name__ = "users"
+    c: ClassVar[FieldProxy] = FieldProxy()
+    id: Optional[int] = None
+    username: str = ""
+    __read_only__ = False
+
+
+class AsyncImplementsButWritable(AsyncActiveRecord, IReadOnlyBehavior):
+    __table_name__ = "users"
+    c: ClassVar[FieldProxy] = FieldProxy()
+    id: Optional[int] = None
+    username: str = ""
+
+
+class AsyncReadOnlyView(AsyncActiveRecord, ReadOnlyMixin):
+    __table_name__ = "users"
+    c: ClassVar[FieldProxy] = FieldProxy()
+    id: Optional[int] = None
+    username: str = ""
+
+    @classmethod
+    def object_type(cls) -> ObjectDeclaration:
+        return ObjectDeclaration.VIEW
+
+
+@pytest.fixture
+async def async_readonly_models(async_user_class):
+    backend = async_user_class.backend()
+    for model in (
+        AsyncReadOnlyUser,
+        AsyncWritableUser,
+        AsyncOptedOutUser,
+        AsyncImplementsButWritable,
+        AsyncReadOnlyView,
+    ):
+        model.__backend__ = backend
+    yield async_user_class
+
+
+class TestAsyncReadonlyDeclaration:
+    """The declaration, not a type test, decides."""
+
+    def test_mixin_satisfies_the_interface(self):
+        assert issubclass(ReadOnlyMixin, IReadOnlyBehavior)
+        assert issubclass(AsyncReadOnlyUser, IReadOnlyBehavior)
+
+    def test_read_only_reports_true(self):
+        assert AsyncReadOnlyUser.read_only() is True
+
+    def test_writable_model_has_no_read_only_at_all(self):
+        assert not hasattr(AsyncWritableUser, "read_only")
+
+    def test_writable_model_is_never_refused(self):
+        AsyncWritableUser.refuse_read_only("save")
+
+    def test_explicit_false_opts_back_out(self):
+        assert AsyncOptedOutUser.read_only() is False
+        AsyncOptedOutUser.refuse_read_only("save")
+
+    def test_interface_implementor_reporting_false_still_writes(self):
+        assert issubclass(AsyncImplementsButWritable, IReadOnlyBehavior)
+        assert AsyncImplementsButWritable.read_only() is False
+        AsyncImplementsButWritable.refuse_read_only("save")
+
+    def test_read_only_is_orthogonal_to_the_primary_key(self):
+        assert AsyncReadOnlyUser.primary_key_columns() == ("id",)
+        assert AsyncWritableUser.primary_key_columns() == ("id",)
+
+    def test_read_only_model_can_declare_a_view_object_type(self):
+        assert AsyncReadOnlyUser.object_type() is ObjectDeclaration.TABLE
+        assert AsyncReadOnlyView.object_type() is ObjectDeclaration.VIEW
+
+
+class TestAsyncReadonlyWriteRefusal:
+    async def test_save_raises(self, async_readonly_models):
+        with pytest.raises(ReadOnlyError, match="save is not allowed on read-only model"):
+            await AsyncReadOnlyUser(username="x").save()
+
+    async def test_delete_raises(self, async_readonly_models):
+        record = AsyncReadOnlyUser(id=1, username="x")
+        record._is_from_db = True
+        with pytest.raises(ReadOnlyError, match="delete is not allowed"):
+            await record.delete()
+
+    def test_error_names_model_and_operation(self):
+        with pytest.raises(ReadOnlyError) as excinfo:
+            AsyncReadOnlyUser.refuse_read_only("bulk_delete")
+        assert excinfo.value.model_name == "AsyncReadOnlyUser"
+        assert excinfo.value.operation == "bulk_delete"
+
+    def test_error_is_a_database_error(self):
+        assert issubclass(ReadOnlyError, DatabaseError)
+
+
+class TestAsyncReadonlyClassLevelRefusal:
+    async def test_bulk_create_raises(self, async_readonly_models):
+        with pytest.raises(ReadOnlyError, match="bulk_create"):
+            await AsyncReadOnlyUser.bulk_create([AsyncReadOnlyUser(username="x")])
+
+    async def test_bulk_update_raises(self, async_readonly_models):
+        with pytest.raises(ReadOnlyError, match="bulk_update"):
+            await AsyncReadOnlyUser.bulk_update([AsyncReadOnlyUser(id=1, username="x")], ["username"])
+
+    async def test_bulk_delete_raises(self, async_readonly_models):
+        with pytest.raises(ReadOnlyError, match="bulk_delete"):
+            await AsyncReadOnlyUser.bulk_delete([AsyncReadOnlyUser(id=1, username="x")])
+
+    async def test_update_all_raises(self, async_readonly_models):
+        query = AsyncReadOnlyUser.query().where(AsyncReadOnlyUser.c.id == 1)
+        with pytest.raises(ReadOnlyError, match="update_all"):
+            await query.update_all({"username": "y"})
+
+    async def test_delete_all_raises(self, async_readonly_models):
+        query = AsyncReadOnlyUser.query().where(AsyncReadOnlyUser.c.id == 1)
+        with pytest.raises(ReadOnlyError, match="delete_all"):
+            await query.delete_all()
+
+
+class TestAsyncReadonlyReadsStillWork:
+    async def test_query_works(self, async_readonly_models):
+        assert isinstance(await AsyncReadOnlyUser.query().all(), list)
+
+    async def test_count_works(self, async_readonly_models):
+        assert await AsyncReadOnlyUser.query().count() >= 0
+
+    async def test_where_works(self, async_readonly_models):
+        assert isinstance(await AsyncReadOnlyUser.query().where(AsyncReadOnlyUser.c.id == 1).all(), list)

From 6df8ac42ff7cd3ec35cfd39814bee7df3bc91892 Mon Sep 17 00:00:00 2001
From: Zhao vistart <i@vistart.me>
Date: Tue, 10 Nov 2026 00:52:53 +0800
Subject: [PATCH 2/3] test(readonly): cover find_one_or_create read and refusal
 paths

---
 .../basic/readonly/test_readonly_models.py    | 28 ++++++++++++++++
 .../readonly/test_readonly_models_async.py    | 32 +++++++++++++++++++
 2 files changed, 60 insertions(+)

diff --git a/src/rhosocial/activerecord/testsuite/feature/basic/readonly/test_readonly_models.py b/src/rhosocial/activerecord/testsuite/feature/basic/readonly/test_readonly_models.py
index 4752e6d..b9f1b08 100644
--- a/src/rhosocial/activerecord/testsuite/feature/basic/readonly/test_readonly_models.py
+++ b/src/rhosocial/activerecord/testsuite/feature/basic/readonly/test_readonly_models.py
@@ -167,3 +167,31 @@ def test_count_works(self, readonly_models):
 
     def test_where_works(self, readonly_models):
         assert isinstance(ReadOnlyUser.query().where(ReadOnlyUser.c.id == 1).all(), list)
+
+
+class TestReadonlyFindOneOrCreate:
+    """``find_one_or_create`` keeps the read path but refuses creation."""
+
+    _EXISTING_USERNAME = "__readonly_find_one_or_create_existing__"
+    _EXISTING_EMAIL = "readonly-find-one-or-create@example.com"
+    _MISSING_USERNAME = "__readonly_find_one_or_create_missing__"
+
+    def test_existing_row_is_returned_without_writing(self, readonly_models):
+        seed = readonly_models(
+            username=self._EXISTING_USERNAME,
+            email=self._EXISTING_EMAIL,
+            age=25,
+        )
+        seed.save()
+        record, created = ReadOnlyUser.find_one_or_create(
+            username=self._EXISTING_USERNAME
+        )
+        assert created is False
+        assert record is not None
+        assert record.id == seed.id
+
+    def test_missing_row_refuses_creation(self, readonly_models):
+        with pytest.raises(
+            ReadOnlyError, match="save is not allowed on read-only model"
+        ):
+            ReadOnlyUser.find_one_or_create(username=self._MISSING_USERNAME)
diff --git a/src/rhosocial/activerecord/testsuite/feature/basic/readonly/test_readonly_models_async.py b/src/rhosocial/activerecord/testsuite/feature/basic/readonly/test_readonly_models_async.py
index 38bf0a1..2459966 100644
--- a/src/rhosocial/activerecord/testsuite/feature/basic/readonly/test_readonly_models_async.py
+++ b/src/rhosocial/activerecord/testsuite/feature/basic/readonly/test_readonly_models_async.py
@@ -160,3 +160,35 @@ async def test_count_works(self, async_readonly_models):
 
     async def test_where_works(self, async_readonly_models):
         assert isinstance(await AsyncReadOnlyUser.query().where(AsyncReadOnlyUser.c.id == 1).all(), list)
+
+
+class TestAsyncReadonlyFindOneOrCreate:
+    """``find_one_or_create`` keeps the read path but refuses creation."""
+
+    _EXISTING_USERNAME = "__readonly_find_one_or_create_existing__"
+    _EXISTING_EMAIL = "readonly-find-one-or-create@example.com"
+    _MISSING_USERNAME = "__readonly_find_one_or_create_missing__"
+
+    async def test_existing_row_is_returned_without_writing(
+        self, async_readonly_models
+    ):
+        seed = async_readonly_models(
+            username=self._EXISTING_USERNAME,
+            email=self._EXISTING_EMAIL,
+            age=25,
+        )
+        await seed.save()
+        record, created = await AsyncReadOnlyUser.find_one_or_create(
+            username=self._EXISTING_USERNAME
+        )
+        assert created is False
+        assert record is not None
+        assert record.id == seed.id
+
+    async def test_missing_row_refuses_creation(self, async_readonly_models):
+        with pytest.raises(
+            ReadOnlyError, match="save is not allowed on read-only model"
+        ):
+            await AsyncReadOnlyUser.find_one_or_create(
+                username=self._MISSING_USERNAME
+            )

From 3a04dad14788c4f81c404eef7eab0d6639531315 Mon Sep 17 00:00:00 2001
From: Zhao vistart <i@vistart.me>
Date: Tue, 10 Nov 2026 01:02:22 +0800
Subject: [PATCH 3/3] chore: add changelog fragment for PR #30

---
 changelog.d/30.added.md | 1 +
 1 file changed, 1 insertion(+)
 create mode 100644 changelog.d/30.added.md

diff --git a/changelog.d/30.added.md b/changelog.d/30.added.md
new file mode 100644
index 0000000..2bb9d4d
--- /dev/null
+++ b/changelog.d/30.added.md
@@ -0,0 +1 @@
+Added sync and async contracts for read-only and keyless models, including the `find_one_or_create` read and refusal paths.