diff --git a/configs/dev.yaml b/configs/dev.yaml index 9ee32ad..028d1af 100644 --- a/configs/dev.yaml +++ b/configs/dev.yaml @@ -11,3 +11,11 @@ S3_ACCESS_KEY: "dev-access" S3_SECRET_KEY: "5ece9238b18dc29d5e0724b3ae577cdf90b8f78b5eaf62ca" S3_BUCKET_NAME: "dev" S3_VERIFY: "true" + +KAFKA_BROKERS: "localhost:9092" +KAFKA_REQUEST_TOPIC: "moderation-request" +KAFKA_RESULT_TOPIC: "moderation-result" +KAFKA_CONSUMER_GROUP_ID: "moderation-consumer-group" +KAFKA_MAX_RETRIES: "3" +KAFKA_RETRY_BACKOFF_MS: "1000" +KAFKA_ENABLE_SSL: false diff --git a/justfile b/justfile index 5d25313..77b40bf 100644 --- a/justfile +++ b/justfile @@ -10,6 +10,10 @@ TMP_DIR := '.proto' SOURCE := 'modservice' OUT_DIR := 'src/' + SOURCE + '/grpc' +MODERATION_PROTO_TAG := 'v0.1.3' +MODERATION_PROTO_NAME := 'moderation.proto' +MODERATION_TMP_DIR := '.proto_moderation' + MKDIR_TOOLS := 'mkdir -p tools' FETCH_COMMON_JUST := 'curl -fsSL ' + COMMON_JUST_URL + ' -o tools/common.just' diff --git a/migrations/20260308000000_add_on_moderation_status.sql b/migrations/20260308000000_add_on_moderation_status.sql new file mode 100644 index 0000000..9458120 --- /dev/null +++ b/migrations/20260308000000_add_on_moderation_status.sql @@ -0,0 +1,22 @@ +-- +goose Up +-- +goose NO TRANSACTION + +-- Add ON_MODERATION to existing mod_status enum +ALTER TYPE mod_status ADD VALUE IF NOT EXISTS 'ON_MODERATION'; + +-- +goose StatementBegin +-- Update existing UPLOADING mods to ON_MODERATION +UPDATE mods SET status = 'ON_MODERATION' WHERE status = 'UPLOADING'; +-- +goose StatementEnd + +-- +goose Down +-- +goose StatementBegin + +-- Revert ON_MODERATION mods back to UPLOADING +UPDATE mods SET status = 'UPLOADING' WHERE status = 'ON_MODERATION'; + +-- Note: PostgreSQL does not support removing enum values directly +-- If rollback is needed, the enum type would need to be recreated +-- For now, we just revert the data changes + +-- +goose StatementEnd diff --git a/pdm.lock b/pdm.lock index 7f0e05d..d5082ec 100644 --- a/pdm.lock +++ b/pdm.lock @@ -5,7 +5,7 @@ groups = ["default", "dev"] strategy = ["inherit_metadata"] lock_version = "4.5.0" -content_hash = "sha256:7d4b4cbaad3dcb3661f2c47f4e53b29584711390ec6ef016f1b925bf24f9a2a6" +content_hash = "sha256:26d30cdb6d5cf685c8bc4e118c5471214345e0d78781b7264b3a136dcd330955" [[metadata.targets]] requires_python = "~=3.13" @@ -359,6 +359,26 @@ files = [ {file = "colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44"}, ] +[[package]] +name = "confluent-kafka" +version = "2.13.2" +requires_python = ">=3.8" +summary = "Confluent's Python client for Apache Kafka" +groups = ["default"] +files = [ + {file = "confluent_kafka-2.13.2-cp313-cp313-macosx_13_0_arm64.whl", hash = "sha256:02702808dd3cfd91f117fbf17181da2a95392967e9f946b1cbdc5589b36e39d1"}, + {file = "confluent_kafka-2.13.2-cp313-cp313-macosx_13_0_x86_64.whl", hash = "sha256:7dc3a2da92638c077bbabb07058f1938078b42a89f0bbfdcb852d4289c2de27e"}, + {file = "confluent_kafka-2.13.2-cp313-cp313-manylinux_2_28_aarch64.whl", hash = "sha256:f3e6d010ad38447a48e0f9fab81edd4d2fd0b5f5a79ab475c30347689e35c6e6"}, + {file = "confluent_kafka-2.13.2-cp313-cp313-manylinux_2_28_x86_64.whl", hash = "sha256:9161865d8246eb77d1c30233a315bdad96145af783981877664532fa212f56be"}, + {file = "confluent_kafka-2.13.2-cp313-cp313-win_amd64.whl", hash = "sha256:931233798306b859f4870ec58e3951a2bd32d14ef29f944f56892851b0aafab0"}, + {file = "confluent_kafka-2.13.2-cp314-cp314-macosx_13_0_arm64.whl", hash = "sha256:9cb0d6820107deca1823d68b96831bd982d0a11c4e6bcf0a12e8040192c48a8f"}, + {file = "confluent_kafka-2.13.2-cp314-cp314-macosx_13_0_x86_64.whl", hash = "sha256:b31d94bca493d84927927d1bdd59e1b6d3d921019a657f99f0c8cc5da8c85311"}, + {file = "confluent_kafka-2.13.2-cp314-cp314-manylinux_2_28_aarch64.whl", hash = "sha256:f09adb42fb898a0b3a88b02e77bee472e93f758258945386c77864016b4e4efc"}, + {file = "confluent_kafka-2.13.2-cp314-cp314-manylinux_2_28_x86_64.whl", hash = "sha256:fa3be1fe231e06b2c7501fa3641b30ea90ea17be79ca89806eef22ff34ed106c"}, + {file = "confluent_kafka-2.13.2-cp314-cp314-win_amd64.whl", hash = "sha256:a8d1e0721de378034ecc928b47238272b56bf20af5dd504233bcb93ce07a38a6"}, + {file = "confluent_kafka-2.13.2.tar.gz", hash = "sha256:619d10d1d77c9821ba913b3e42a33ade7f889f3573c7f3c17b57c3056e3310f5"}, +] + [[package]] name = "coverage" version = "7.11.0" @@ -707,6 +727,21 @@ files = [ {file = "grpcio_tools-1.75.1.tar.gz", hash = "sha256:bb78960cf3d58941e1fec70cbdaccf255918beed13c34112a6915a6d8facebd1"}, ] +[[package]] +name = "hypothesis" +version = "6.151.9" +requires_python = ">=3.10" +summary = "The property-based testing library for Python" +groups = ["dev"] +dependencies = [ + "exceptiongroup>=1.0.0; python_version < \"3.11\"", + "sortedcontainers<3.0.0,>=2.1.0", +] +files = [ + {file = "hypothesis-6.151.9-py3-none-any.whl", hash = "sha256:7b7220585c67759b1b1ef839b1e6e9e3d82ed468cfc1ece43c67184848d7edd9"}, + {file = "hypothesis-6.151.9.tar.gz", hash = "sha256:2f284428dda6c3c48c580de0e18470ff9c7f5ef628a647ee8002f38c3f9097ca"}, +] + [[package]] name = "idna" version = "3.10" @@ -1360,6 +1395,16 @@ files = [ {file = "sniffio-1.3.1.tar.gz", hash = "sha256:f4324edc670a0f49750a81b895f35c3adb843cca46f0530f79fc1babb23789dc"}, ] +[[package]] +name = "sortedcontainers" +version = "2.4.0" +summary = "Sorted Containers -- Sorted List, Sorted Dict, Sorted Set" +groups = ["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"}, +] + [[package]] name = "types-aioboto3" version = "15.4.0" diff --git a/pyproject.toml b/pyproject.toml index 08d999a..fedc3e8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -124,6 +124,8 @@ dependencies = [ "aioboto3>=13.0.0", "python-dotenv>=1.0.0", "asyncpg>=0.30.0", + "confluent-kafka>=2.12.2", + "protobuf==6.32.1", ] requires-python = ">=3.13" readme = "README.md" @@ -144,6 +146,7 @@ dev = [ "pytest-mock==3.15.1", "pytest-faker==2.0.0", "Faker==37.11.0", + "hypothesis>=6.151.9", "black==25.9.0", "isort==6.0.1", "flake8==7.3.0", diff --git a/src/modservice/constants.py b/src/modservice/constants.py index b47af9d..0377f4f 100644 --- a/src/modservice/constants.py +++ b/src/modservice/constants.py @@ -1,4 +1,5 @@ STATUS_UPLOADING = "UPLOADING" +STATUS_ON_MODERATION = "ON_MODERATION" STATUS_UPLOADED = "UPLOADED" STATUS_BANNED = "BANNED" STATUS_HIDDEN = "HIDDEN" diff --git a/src/modservice/grpc/__init__.py b/src/modservice/grpc/__init__.py new file mode 100644 index 0000000..304105f --- /dev/null +++ b/src/modservice/grpc/__init__.py @@ -0,0 +1 @@ +# Generated gRPC files diff --git a/src/modservice/grpc/moderation_pb2.py b/src/modservice/grpc/moderation_pb2.py new file mode 100644 index 0000000..8860587 --- /dev/null +++ b/src/modservice/grpc/moderation_pb2.py @@ -0,0 +1,42 @@ +# -*- coding: utf-8 -*- +# Generated by the protocol buffer compiler. DO NOT EDIT! +# NO CHECKED-IN PROTOBUF GENCODE +# source: moderation.proto +# Protobuf Python Version: 6.31.1 +"""Generated protocol buffer code.""" +from google.protobuf import descriptor as _descriptor +from google.protobuf import descriptor_pool as _descriptor_pool +from google.protobuf import runtime_version as _runtime_version +from google.protobuf import symbol_database as _symbol_database +from google.protobuf.internal import builder as _builder +_runtime_version.ValidateProtobufRuntimeVersion( + _runtime_version.Domain.PUBLIC, + 6, + 31, + 1, + '', + 'moderation.proto' +) +# @@protoc_insertion_point(imports) + +_sym_db = _symbol_database.Default() + + + + +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x10moderation.proto\x12\nmoderation\"W\n\x15ModerateObjectRequest\x12\n\n\x02id\x18\x01 \x01(\x03\x12$\n\x04type\x18\x02 \x01(\x0e\x32\x16.moderation.ObjectType\x12\x0c\n\x04text\x18\x03 \x01(\t\")\n\x16ModerateObjectResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08*\x83\x01\n\nObjectType\x12\x1b\n\x17OBJECT_TYPE_UNSPECIFIED\x10\x00\x12\x1f\n\x1bOBJECT_TYPE_MOD_DESCRIPTION\x10\x01\x12\x1c\n\x18OBJECT_TYPE_COMMENT_TEXT\x10\x02\x12\x19\n\x15OBJECT_TYPE_USER_NAME\x10\x03\x32l\n\x11ModerationService\x12W\n\x0eModerateObject\x12!.moderation.ModerateObjectRequest\x1a\".moderation.ModerateObjectResponseb\x06proto3') + +_globals = globals() +_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) +_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'moderation_pb2', _globals) +if not _descriptor._USE_C_DESCRIPTORS: + DESCRIPTOR._loaded_options = None + _globals['_OBJECTTYPE']._serialized_start=165 + _globals['_OBJECTTYPE']._serialized_end=296 + _globals['_MODERATEOBJECTREQUEST']._serialized_start=32 + _globals['_MODERATEOBJECTREQUEST']._serialized_end=119 + _globals['_MODERATEOBJECTRESPONSE']._serialized_start=121 + _globals['_MODERATEOBJECTRESPONSE']._serialized_end=162 + _globals['_MODERATIONSERVICE']._serialized_start=298 + _globals['_MODERATIONSERVICE']._serialized_end=406 +# @@protoc_insertion_point(module_scope) diff --git a/src/modservice/grpc/moderation_pb2.pyi b/src/modservice/grpc/moderation_pb2.pyi new file mode 100644 index 0000000..107cea2 --- /dev/null +++ b/src/modservice/grpc/moderation_pb2.pyi @@ -0,0 +1,33 @@ +from google.protobuf.internal import enum_type_wrapper as _enum_type_wrapper +from google.protobuf import descriptor as _descriptor +from google.protobuf import message as _message +from typing import ClassVar as _ClassVar, Optional as _Optional, Union as _Union + +DESCRIPTOR: _descriptor.FileDescriptor + +class ObjectType(int, metaclass=_enum_type_wrapper.EnumTypeWrapper): + __slots__ = () + OBJECT_TYPE_UNSPECIFIED: _ClassVar[ObjectType] + OBJECT_TYPE_MOD_DESCRIPTION: _ClassVar[ObjectType] + OBJECT_TYPE_COMMENT_TEXT: _ClassVar[ObjectType] + OBJECT_TYPE_USER_NAME: _ClassVar[ObjectType] +OBJECT_TYPE_UNSPECIFIED: ObjectType +OBJECT_TYPE_MOD_DESCRIPTION: ObjectType +OBJECT_TYPE_COMMENT_TEXT: ObjectType +OBJECT_TYPE_USER_NAME: ObjectType + +class ModerateObjectRequest(_message.Message): + __slots__ = ("id", "type", "text") + ID_FIELD_NUMBER: _ClassVar[int] + TYPE_FIELD_NUMBER: _ClassVar[int] + TEXT_FIELD_NUMBER: _ClassVar[int] + id: int + type: ObjectType + text: str + def __init__(self, id: _Optional[int] = ..., type: _Optional[_Union[ObjectType, str]] = ..., text: _Optional[str] = ...) -> None: ... + +class ModerateObjectResponse(_message.Message): + __slots__ = ("success",) + SUCCESS_FIELD_NUMBER: _ClassVar[int] + success: bool + def __init__(self, success: bool = ...) -> None: ... diff --git a/src/modservice/grpc/moderation_pb2_grpc.py b/src/modservice/grpc/moderation_pb2_grpc.py new file mode 100644 index 0000000..ed857d0 --- /dev/null +++ b/src/modservice/grpc/moderation_pb2_grpc.py @@ -0,0 +1,97 @@ +# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT! +"""Client and server classes corresponding to protobuf-defined services.""" +import grpc +import warnings + +from . import moderation_pb2 as moderation__pb2 + +GRPC_GENERATED_VERSION = '1.75.1' +GRPC_VERSION = grpc.__version__ +_version_not_supported = False + +try: + from grpc._utilities import first_version_is_lower + _version_not_supported = first_version_is_lower(GRPC_VERSION, GRPC_GENERATED_VERSION) +except ImportError: + _version_not_supported = True + +if _version_not_supported: + raise RuntimeError( + f'The grpc package installed is at version {GRPC_VERSION},' + + f' but the generated code in moderation_pb2_grpc.py depends on' + + f' grpcio>={GRPC_GENERATED_VERSION}.' + + f' Please upgrade your grpc module to grpcio>={GRPC_GENERATED_VERSION}' + + f' or downgrade your generated code using grpcio-tools<={GRPC_VERSION}.' + ) + + +class ModerationServiceStub(object): + """Missing associated documentation comment in .proto file.""" + + def __init__(self, channel): + """Constructor. + + Args: + channel: A grpc.Channel. + """ + self.ModerateObject = channel.unary_unary( + '/moderation.ModerationService/ModerateObject', + request_serializer=moderation__pb2.ModerateObjectRequest.SerializeToString, + response_deserializer=moderation__pb2.ModerateObjectResponse.FromString, + _registered_method=True) + + +class ModerationServiceServicer(object): + """Missing associated documentation comment in .proto file.""" + + def ModerateObject(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + + +def add_ModerationServiceServicer_to_server(servicer, server): + rpc_method_handlers = { + 'ModerateObject': grpc.unary_unary_rpc_method_handler( + servicer.ModerateObject, + request_deserializer=moderation__pb2.ModerateObjectRequest.FromString, + response_serializer=moderation__pb2.ModerateObjectResponse.SerializeToString, + ), + } + generic_handler = grpc.method_handlers_generic_handler( + 'moderation.ModerationService', rpc_method_handlers) + server.add_generic_rpc_handlers((generic_handler,)) + server.add_registered_method_handlers('moderation.ModerationService', rpc_method_handlers) + + + # This class is part of an EXPERIMENTAL API. +class ModerationService(object): + """Missing associated documentation comment in .proto file.""" + + @staticmethod + def ModerateObject(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_unary( + request, + target, + '/moderation.ModerationService/ModerateObject', + moderation__pb2.ModerateObjectRequest.SerializeToString, + moderation__pb2.ModerateObjectResponse.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True) diff --git a/src/modservice/handler/get_mods.py b/src/modservice/handler/get_mods.py index dd61c18..85af35c 100644 --- a/src/modservice/handler/get_mods.py +++ b/src/modservice/handler/get_mods.py @@ -1,10 +1,11 @@ import grpc -from modservice.constants import STATUS_BANNED, STATUS_HIDDEN, STATUS_UPLOADED +from modservice.constants import STATUS_BANNED, STATUS_HIDDEN, STATUS_ON_MODERATION, STATUS_UPLOADED from modservice.grpc import mod_pb2 from modservice.service.service import ModService STATUS_TO_PROTO: dict[str, mod_pb2.ModStatus] = { + STATUS_ON_MODERATION: mod_pb2.ModStatus.MOD_STATUS_UNSPECIFIED, STATUS_UPLOADED: mod_pb2.ModStatus.MOD_STATUS_UPLOADED, STATUS_BANNED: mod_pb2.ModStatus.MOD_STATUS_BANNED, STATUS_HIDDEN: mod_pb2.ModStatus.MOD_STATUS_HIDDEN, diff --git a/src/modservice/kafka/__init__.py b/src/modservice/kafka/__init__.py new file mode 100644 index 0000000..f9c35c8 --- /dev/null +++ b/src/modservice/kafka/__init__.py @@ -0,0 +1,6 @@ +"""Kafka components for moderation integration.""" + +from modservice.kafka.consumer import KafkaConsumer +from modservice.kafka.producer import KafkaProducer + +__all__ = ["KafkaConsumer", "KafkaProducer"] diff --git a/src/modservice/kafka/config.py b/src/modservice/kafka/config.py new file mode 100644 index 0000000..89f5517 --- /dev/null +++ b/src/modservice/kafka/config.py @@ -0,0 +1,37 @@ +import os + + +class KafkaConfig: + + def __init__(self): + self.brokers = os.getenv('KAFKA_BROKERS', 'localhost:9092') + self.request_topic = os.getenv('KAFKA_REQUEST_TOPIC', 'moderation-request') + self.result_topic = os.getenv('KAFKA_RESULT', 'moderation-result') + self.consumer_group_id = os.getenv('KAFKA_CONSUMER_GROUP_ID', 'mod-service-group') + self.max_retries = int(os.getenv('KAFKA_MAX_RETRIES', '3')) + self.retry_backoff_ms = int(os.getenv('KAFKA_RETRY_BACKOFF_MS', '1000')) + self.enable_ssl = os.getenv('KAFKA_ENABLE_SSL', 'false').lower() == 'true' + + def get_producer_config(self): + config = { + 'bootstrap.servers': self.brokers, + 'client.id': 'mod_service_producer', + 'acks': 'all', + 'retries': self.max_retries, + 'retry.backoff.ms': self.retry_backoff_ms, + 'enable.idempotence': True, + 'compression.type': 'none' + } + return config + + def get_consumer_config(self): + config = { + 'bootstrap.servers': self.brokers, + 'group.id': self.consumer_group_id, + 'auto.offset.reset': 'earliest', + 'enable.auto.commit': True, + 'auto.commit.interval.ms': 1000, + 'session.timeout.ms': 30000, + 'heartbeat.interval.ms': 10000 + } + return config diff --git a/src/modservice/kafka/consumer.py b/src/modservice/kafka/consumer.py new file mode 100644 index 0000000..9f2e255 --- /dev/null +++ b/src/modservice/kafka/consumer.py @@ -0,0 +1,108 @@ +"""Kafka consumer for receiving moderation responses.""" + +from confluent_kafka import Consumer, KafkaException, KafkaError +import logging +import threading +from .config import KafkaConfig +from modservice.grpc import moderation_pb2 + +logger = logging.getLogger(__name__) + + +class KafkaConsumer: + + def __init__(self, config: KafkaConfig, callback): + + self.config = config + self.callback = callback + self.consumer = Consumer(config.get_consumer_config()) + self.topic = config.result_topic + self.running = False + self.consumer_thread = None + + self.consumer.subscribe([self.topic]) + logger.info(f"KafkaConsumer subscribed to topic: {self.topic}") + + def start(self): + + if self.running: + logger.warning("Consumer already running") + return + + self.running = True + self.consumer_thread = threading.Thread(target=self._consume_loop, daemon=True) + self.consumer_thread.start() + logger.info("KafkaConsumer started") + + def stop(self): + + if not self.running: + return + + self.running = False + + if self.consumer_thread and self.consumer_thread.is_alive(): + self.consumer_thread.join(timeout=5.0) + + self.consumer.close() + logger.info("KafkaConsumer stopped") + + def _consume_loop(self): + + logger.info("Consumer loop started") + + while self.running: + try: + msg = self.consumer.poll(timeout=1.0) + + if msg is None: + continue + if msg.error(): + if msg.error().code() == KafkaError._PARTITION_EOF: + logger.debug(f"Reached end of partition for topic {msg.topic()}") + elif msg.error().code() == KafkaError._TIMED_OUT: + logger.debug("Timeout") + else: + logger.error(f"Consumer error: {msg.error()}") + continue + + self._process_message(msg) + + except KafkaException as e: + logger.error(f"Kafka exception in consume loop: {e}") + except Exception as e: + logger.error(f"Unexpected error in consume loop: {e}") + + logger.info("Consumer loop ended") + + def _process_message(self, msg): + + try: + object_id = 0 + if msg.key() is not None: + try: + key_str = msg.key().decode('utf-8') + object_id = int(key_str) + except ValueError as e: + logger.error(f"Failed to parse message key to request ID: {e}") + return + + response = moderation_pb2.ModerateObjectResponse() + response.ParseFromString(msg.value()) + + logger.info( + f"Received moderation response: object_id={object_id}, " + f"success={response.success}" + ) + + if self.callback: + try: + self.callback(response, object_id) + except Exception as e: + logger.error(f"Error in callback: {e}") + + except Exception as e: + logger.error(f"Failed to process message: {e}") + + def is_running(self) -> bool: + return self.running diff --git a/src/modservice/kafka/producer.py b/src/modservice/kafka/producer.py new file mode 100644 index 0000000..f25618e --- /dev/null +++ b/src/modservice/kafka/producer.py @@ -0,0 +1,72 @@ +"""Kafka producer for sending moderation requests.""" + +from confluent_kafka import Producer, KafkaException +import logging +from .config import KafkaConfig +from modservice.grpc import moderation_pb2 + +logger = logging.getLogger(__name__) + + +class KafkaProducer: + + def __init__(self, config: KafkaConfig): + self.config = config + self.producer = Producer(config.get_producer_config()) + self.topic = config.request_topic + + logger.info(f"KafkaProducer initialized for topic: {self.topic}") + + def send_moderation_request(self, request: moderation_pb2.ModerateObjectRequest, retries: int = 3) -> bool: + for _ in range(retries): + try: + serialized_request = request.SerializeToString() + + key = str(request.id).encode('utf-8') + + self.producer.produce( + topic=self.topic, + value=serialized_request, + key=key, + partition=-1, + callback=self._delivery_callback + ) + + self.producer.poll(0) + + logger.debug(f"Moderation request queued: ID={request.id}, text_length={len(request.text)}") + return True + + except BufferError as e: + logger.error(f"Producer buffer full, message not queued: {e}") + return False + except KafkaException as e: + logger.error(f"Failed to produce message: {e}") + return False + except Exception as e: + logger.error(f"Unexpected error producing message: {e}") + return False + return False + + def _delivery_callback(self, err, msg): + + if err is not None: + logger.error(f"Message delivery failed: {err}") + else: + logger.debug( + f"Message delivered: topic={msg.topic()}, " + f"partition={msg.partition()}, offset={msg.offset()}" + ) + + def flush(self, timeout: float = 10.0): + + remaining = self.producer.flush(timeout) + + if remaining > 0: + logger.warning(f"Flush incomplete: {remaining} messages still pending") + else: + logger.debug("All messages flushed successfully") + + def __del__(self): + if hasattr(self, 'producer'): + self.flush() diff --git a/src/modservice/moderation/moderation_service.py b/src/modservice/moderation/moderation_service.py new file mode 100644 index 0000000..4b2e50d --- /dev/null +++ b/src/modservice/moderation/moderation_service.py @@ -0,0 +1,91 @@ +"""Moderation service coordination layer.""" + +import logging +import asyncio + +from modservice.repository.repository import ModRepository +from modservice.kafka.producer import KafkaProducer +from modservice.kafka.consumer import KafkaConsumer +from modservice.kafka.config import KafkaConfig +from modservice.grpc import moderation_pb2 + +logger = logging.getLogger(__name__) + + +class ModerationService: + + def __init__(self, repo: ModRepository, loop: asyncio.AbstractEventLoop): + self.config = KafkaConfig() + self._repo = repo + self._loop = loop + + self.producer = KafkaProducer(self.config) + self.consumer = KafkaConsumer( + self.config, + callback=self._handle_moderation_response + ) + + self.consumer.start() + + logger.info("ModerationService initialized") + + def send_moderation_request(self, mod_id: int, title: str, description: str) -> bool: + try: + request = moderation_pb2.ModerateObjectRequest() + request.id = mod_id + request.text = f"{title} {description}" + request.type = moderation_pb2.OBJECT_TYPE_MOD_DESCRIPTION + + success = self.producer.send_moderation_request(request) + + if success: + logger.info(f"Moderation request sent: mod_id={mod_id}") + else: + logger.error(f"Failed to send moderation request: mod_id={mod_id}") + + return success + + except Exception as e: + logger.error(f"Error requesting moderation: {e}") + return False + + def _handle_moderation_response(self, response: moderation_pb2.ModerateObjectResponse, object_id: int): + + try: + approved = response.success + + logger.info( + f"Moderation result received: mod_id={object_id}, " + f"approved={approved}" + ) + + self._update_mod_status(object_id, approved) + + except Exception as e: + logger.error(f"Error handling moderation response: {e}") + + def _update_mod_status(self, mod_id: int, approved: bool) -> None: + new_status = "UPLOADED" if approved else "HIDDEN" + + logger.info(f"Updating mod status: mod_id={mod_id}, approved={approved}, new_status={new_status}") + + coroutine = self._repo.set_status(mod_id, new_status) + + future = asyncio.run_coroutine_threadsafe(coroutine, self._loop) + + def _log_result(future: asyncio.Future) -> None: + try: + success = future.result() + if not success: + logger.error(f"Failed to update mod status: mod_id={mod_id}") + else: + logger.info(f"Successfully updated mod status: mod_id={mod_id}, status={new_status}") + except Exception as e: + logger.error(f"Error updating mod status: {e}") + + future.add_done_callback(_log_result) + + def shutdown(self): + self.consumer.stop() + self.producer.flush() + logger.info("ModerationService shutdown complete") diff --git a/src/modservice/repository/create_mod.py b/src/modservice/repository/create_mod.py index b7e7368..0e90b83 100644 --- a/src/modservice/repository/create_mod.py +++ b/src/modservice/repository/create_mod.py @@ -11,7 +11,7 @@ async def create_mod( mod_id: int = await conn.fetchval( """ INSERT INTO mods (author_id, title, description, version, status, created_at) - VALUES ($1, $2, $3, $4, 'UPLOADING', NOW()) + VALUES ($1, $2, $3, $4, 'ON_MODERATION', NOW()) RETURNING id """, author_id, diff --git a/src/modservice/repository/set_status.py b/src/modservice/repository/set_status.py index 3e01070..3f7e53f 100644 --- a/src/modservice/repository/set_status.py +++ b/src/modservice/repository/set_status.py @@ -2,18 +2,15 @@ async def set_status(db_pool: Pool, mod_id: int, status: str) -> bool: - try: - async with db_pool.acquire() as conn: - result = await conn.execute( - """ - UPDATE mods - SET status = $1 - WHERE id = $2 - """, - status, - mod_id, - ) - rows_affected = int(result.split()[-1]) if result else 0 - return rows_affected > 0 - except Exception: - return False + async with db_pool.acquire() as conn: + result = await conn.execute( + """ + UPDATE mods + SET status = $1 + WHERE id = $2 + """, + status, + mod_id, + ) + rows_affected = int(result.split()[-1]) if result else 0 + return rows_affected > 0 diff --git a/src/modservice/server.py b/src/modservice/server.py index bd74bef..f83ba8d 100644 --- a/src/modservice/server.py +++ b/src/modservice/server.py @@ -1,13 +1,12 @@ import asyncio import logging from concurrent import futures - import asyncpg import grpc from grpc_reflection.v1alpha import reflection - from modservice.grpc import mod_pb2, mod_pb2_grpc from modservice.handler.handler import ModHandler +from modservice.moderation.moderation_service import ModerationService from modservice.repository.repository import ModRepository from modservice.s3_client import S3Client from modservice.service.s3_service import S3Service @@ -35,26 +34,32 @@ async def serve() -> None: ) s3_service = S3Service(s3_client) - repo = ModRepository(db_pool) - service = ModService(repo, s3_service) + + loop = asyncio.get_event_loop() + moderation_service = ModerationService(repo, loop) + + service = ModService(repo, s3_service, moderation_service) handler = ModHandler(service) server = grpc.aio.server(futures.ThreadPoolExecutor(max_workers=5)) mod_pb2_grpc.add_ModServiceServicer_to_server( handler, server - ) # type: ignore[no-untyped-call] + ) SERVICE_NAMES = ( mod_pb2.DESCRIPTOR.services_by_name["ModService"].full_name, reflection.SERVICE_NAME, ) reflection.enable_server_reflection(SERVICE_NAMES, server) - server.add_insecure_port(f"{settings.host}:{settings.port}") await server.start() logger.info(f"gRPC server listening on {settings.host}:{settings.port}") - await server.wait_for_termination() + + try: + await server.wait_for_termination() + finally: + moderation_service.shutdown() def main() -> None: @@ -62,4 +67,4 @@ def main() -> None: if __name__ == "__main__": - main() + main() \ No newline at end of file diff --git a/src/modservice/service/service.py b/src/modservice/service/service.py index a7c8ae9..4b4a467 100644 --- a/src/modservice/service/service.py +++ b/src/modservice/service/service.py @@ -1,5 +1,6 @@ from typing import Any +from modservice.moderation.moderation_service import ModerationService from modservice.repository.repository import ModRepository from modservice.service.create_mod import create_mod as _create_mod from modservice.service.get_mods import get_mods as _get_mods @@ -8,14 +9,20 @@ class ModService: - def __init__(self, repo: ModRepository, s3_service: S3Service) -> None: + def __init__( + self, + repo: ModRepository, + s3_service: S3Service, + moderation_service: ModerationService, + ) -> None: self._repo = repo self._s3_service = s3_service + self._moderation_service = moderation_service async def create_mod( self, title: str, author_id: int, description: str ) -> tuple[int, str, str]: - return await _create_mod( + result = await _create_mod( self._repo, self._s3_service, title, @@ -23,6 +30,14 @@ async def create_mod( description, ) + # Send moderation request after mod is created (fire-and-forget) + mod_id = result[0] + self._moderation_service.send_moderation_request( + mod_id, title, description + ) + + return result + async def generate_s3_key( self, author_id: int, filename: str, title: str | None = None ) -> str: diff --git a/src/modservice/settings.py b/src/modservice/settings.py index 7ad99db..f8f1199 100644 --- a/src/modservice/settings.py +++ b/src/modservice/settings.py @@ -1,6 +1,6 @@ import logging -from pydantic import Field +from pydantic import Field, field_validator from pydantic_settings import BaseSettings, SettingsConfigDict @@ -23,6 +23,33 @@ class Settings(BaseSettings): s3_bucket_name: str = Field(validation_alias="S3_BUCKET_NAME") s3_verify: bool = Field(validation_alias="S3_VERIFY") + kafka_brokers: str = Field( + validation_alias="KAFKA_BROKERS" + ) + kafka_request_topic: str = Field( + validation_alias="KAFKA_REQUEST_TOPIC" + ) + kafka_result_topic: str = Field( + validation_alias="KAFKA_RESULT_TOPIC" + ) + kafka_consumer_group_id: str = Field( + validation_alias="KAFKA_CONSUMER_GROUP_ID" + ) + + @field_validator( + "kafka_brokers", + "kafka_request_topic", + "kafka_result_topic", + "kafka_consumer_group_id", + ) + @classmethod + def validate_non_empty_string(cls, v: str) -> str: + if not v or not v.strip(): + raise ValueError( + "Kafka configuration fields must be non-empty strings" + ) + return v + def configure_logging(self) -> None: logging.basicConfig( level=self.log_level, diff --git a/tests/config/__init__.py b/tests/config/__init__.py new file mode 100644 index 0000000..5c72ed8 --- /dev/null +++ b/tests/config/__init__.py @@ -0,0 +1 @@ +"""Configuration tests package.""" diff --git a/tests/config/test_settings_validation.py b/tests/config/test_settings_validation.py new file mode 100644 index 0000000..0791555 --- /dev/null +++ b/tests/config/test_settings_validation.py @@ -0,0 +1,194 @@ +"""Property-based tests for Settings configuration validation. + +**Feature: mod-kafka-moderation** +""" + +import pytest +from hypothesis import given, settings +from hypothesis import strategies as st +from pydantic import ValidationError + +from modservice.settings import Settings + + +class TestSettingsConfigurationValidation: + """Property-based tests for Settings configuration validation. + + **Validates: Requirements 1.4** + """ + + @given( + kafka_bootstrap_servers=st.one_of( + st.just(""), + st.just(" "), + st.none(), + ), + kafka_moderation_request_topic=st.text(min_size=1, max_size=50), + kafka_moderation_response_topic=st.text(min_size=1, max_size=50), + kafka_consumer_group_id=st.text(min_size=1, max_size=50), + ) + @settings(max_examples=100) + def test_property_9_invalid_bootstrap_servers_raises_validation_error( + self, + kafka_bootstrap_servers: str | None, + kafka_moderation_request_topic: str, + kafka_moderation_response_topic: str, + kafka_consumer_group_id: str, + ) -> None: + """Property 9: Settings with invalid Kafka config should fail validation. + + **Feature: mod-kafka-moderation, Property 9: For any Settings + instance with invalid Kafka configuration, validation should fail** + + Test that empty or None bootstrap_servers raises validation error. + """ + with pytest.raises((ValidationError, ValueError)): + Settings( + host="localhost", + port=50051, + database_url="postgresql://test", + log_level="INFO", + log_format="%(message)s", + log_datefmt="%Y-%m-%d", + s3_api_endpoint="http://localhost:9000", + s3_access_key="test", + s3_secret_key="test", + s3_bucket_name="test", + s3_verify=False, + kafka_bootstrap_servers=kafka_bootstrap_servers, + kafka_moderation_request_topic=kafka_moderation_request_topic, + kafka_moderation_response_topic=kafka_moderation_response_topic, + kafka_consumer_group_id=kafka_consumer_group_id, + ) + + @given( + kafka_bootstrap_servers=st.text(min_size=1, max_size=50), + kafka_moderation_request_topic=st.one_of( + st.just(""), + st.just(" "), + st.none(), + ), + kafka_moderation_response_topic=st.text(min_size=1, max_size=50), + kafka_consumer_group_id=st.text(min_size=1, max_size=50), + ) + @settings(max_examples=100) + def test_property_9_invalid_request_topic_raises_validation_error( + self, + kafka_bootstrap_servers: str, + kafka_moderation_request_topic: str | None, + kafka_moderation_response_topic: str, + kafka_consumer_group_id: str, + ) -> None: + """Property 9: Settings with invalid Kafka config should fail validation. + + **Feature: mod-kafka-moderation, Property 9: For any Settings + instance with invalid Kafka configuration, validation should fail** + + Test that empty or None request topic raises validation error. + """ + with pytest.raises((ValidationError, ValueError)): + Settings( + host="localhost", + port=50051, + database_url="postgresql://test", + log_level="INFO", + log_format="%(message)s", + log_datefmt="%Y-%m-%d", + s3_api_endpoint="http://localhost:9000", + s3_access_key="test", + s3_secret_key="test", + s3_bucket_name="test", + s3_verify=False, + kafka_bootstrap_servers=kafka_bootstrap_servers, + kafka_moderation_request_topic=kafka_moderation_request_topic, + kafka_moderation_response_topic=kafka_moderation_response_topic, + kafka_consumer_group_id=kafka_consumer_group_id, + ) + + @given( + kafka_bootstrap_servers=st.text(min_size=1, max_size=50), + kafka_moderation_request_topic=st.text(min_size=1, max_size=50), + kafka_moderation_response_topic=st.one_of( + st.just(""), + st.just(" "), + st.none(), + ), + kafka_consumer_group_id=st.text(min_size=1, max_size=50), + ) + @settings(max_examples=100) + def test_property_9_invalid_response_topic_raises_validation_error( + self, + kafka_bootstrap_servers: str, + kafka_moderation_request_topic: str, + kafka_moderation_response_topic: str | None, + kafka_consumer_group_id: str, + ) -> None: + """Property 9: Settings with invalid Kafka config should fail validation. + + **Feature: mod-kafka-moderation, Property 9: For any Settings + instance with invalid Kafka configuration, validation should fail** + + Test that empty or None response topic raises validation error. + """ + with pytest.raises((ValidationError, ValueError)): + Settings( + host="localhost", + port=50051, + database_url="postgresql://test", + log_level="INFO", + log_format="%(message)s", + log_datefmt="%Y-%m-%d", + s3_api_endpoint="http://localhost:9000", + s3_access_key="test", + s3_secret_key="test", + s3_bucket_name="test", + s3_verify=False, + kafka_bootstrap_servers=kafka_bootstrap_servers, + kafka_moderation_request_topic=kafka_moderation_request_topic, + kafka_moderation_response_topic=kafka_moderation_response_topic, + kafka_consumer_group_id=kafka_consumer_group_id, + ) + + @given( + kafka_bootstrap_servers=st.text(min_size=1, max_size=50), + kafka_moderation_request_topic=st.text(min_size=1, max_size=50), + kafka_moderation_response_topic=st.text(min_size=1, max_size=50), + kafka_consumer_group_id=st.one_of( + st.just(""), + st.just(" "), + st.none(), + ), + ) + @settings(max_examples=100) + def test_property_9_invalid_consumer_group_id_raises_validation_error( + self, + kafka_bootstrap_servers: str, + kafka_moderation_request_topic: str, + kafka_moderation_response_topic: str, + kafka_consumer_group_id: str | None, + ) -> None: + """Property 9: Settings with invalid Kafka config should fail validation. + + **Feature: mod-kafka-moderation, Property 9: For any Settings + instance with invalid Kafka configuration, validation should fail** + + Test that empty or None consumer group ID raises validation error. + """ + with pytest.raises((ValidationError, ValueError)): + Settings( + host="localhost", + port=50051, + database_url="postgresql://test", + log_level="INFO", + log_format="%(message)s", + log_datefmt="%Y-%m-%d", + s3_api_endpoint="http://localhost:9000", + s3_access_key="test", + s3_secret_key="test", + s3_bucket_name="test", + s3_verify=False, + kafka_bootstrap_servers=kafka_bootstrap_servers, + kafka_moderation_request_topic=kafka_moderation_request_topic, + kafka_moderation_response_topic=kafka_moderation_response_topic, + kafka_consumer_group_id=kafka_consumer_group_id, + ) diff --git a/tests/kafka/__init__.py b/tests/kafka/__init__.py new file mode 100644 index 0000000..d1a9ac7 --- /dev/null +++ b/tests/kafka/__init__.py @@ -0,0 +1 @@ +# Kafka tests diff --git a/tests/kafka/test_consumer.py b/tests/kafka/test_consumer.py new file mode 100644 index 0000000..89d5899 --- /dev/null +++ b/tests/kafka/test_consumer.py @@ -0,0 +1,231 @@ +"""Unit tests for Kafka consumer.""" + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from modservice.grpc.moderation_pb2 import ModerateObjectResponse +from modservice.kafka.consumer import KafkaConsumer + + +class TestKafkaConsumer: + """Test suite for KafkaConsumer class.""" + + @pytest.fixture + def mock_moderation_service(self) -> AsyncMock: + """Create a mock ModerationService for testing.""" + service = AsyncMock() + service.process_moderation_response = AsyncMock() + return service + + @pytest.fixture + def consumer(self, mock_moderation_service: AsyncMock) -> KafkaConsumer: + """Create a KafkaConsumer instance for testing.""" + return KafkaConsumer( + bootstrap_servers="localhost:9092", + response_topic="moderation.responses", + group_id="test-group", + moderation_service=mock_moderation_service, + ) + + @pytest.mark.asyncio + async def test_start_initializes_consumer( + self, consumer: KafkaConsumer + ) -> None: + """Test that start() initializes the AIOKafkaConsumer connection.""" + with patch( + "modservice.kafka.consumer.AIOKafkaConsumer" + ) as mock_consumer_class: + mock_consumer_instance = AsyncMock() + mock_consumer_class.return_value = mock_consumer_instance + + await consumer.start() + + mock_consumer_class.assert_called_once_with( + "moderation.responses", + bootstrap_servers="localhost:9092", + group_id="test-group", + ) + mock_consumer_instance.start.assert_called_once() + + @pytest.mark.asyncio + async def test_stop_closes_consumer(self, consumer: KafkaConsumer) -> None: + """Test that stop() closes the consumer connection.""" + with patch( + "modservice.kafka.consumer.AIOKafkaConsumer" + ) as mock_consumer_class: + mock_consumer_instance = AsyncMock() + mock_consumer_class.return_value = mock_consumer_instance + + await consumer.start() + await consumer.stop() + + mock_consumer_instance.stop.assert_called_once() + + @pytest.mark.asyncio + async def test_stop_without_start_does_nothing( + self, consumer: KafkaConsumer + ) -> None: + """Test that stop() does nothing if consumer was never started.""" + # Should not raise an exception + await consumer.stop() + + @pytest.mark.asyncio + async def test_consume_messages_processes_response( + self, consumer: KafkaConsumer, mock_moderation_service: AsyncMock + ) -> None: + """Test that consume_messages processes a valid moderation response.""" + with patch( + "modservice.kafka.consumer.AIOKafkaConsumer" + ) as mock_consumer_class: + # Create a valid moderation response + response = ModerateObjectResponse(success=True) + message_bytes = response.SerializeToString() + + # Create message key (object_id as big-endian bytes) + object_id = 123 + object_id_bytes = object_id.to_bytes(8, byteorder="big") + + # Create mock message + mock_message = MagicMock() + mock_message.value = message_bytes + mock_message.key = object_id_bytes + + # Mock the consumer to yield one message then stop + async def mock_aiter(self): # noqa: RUF029 + yield mock_message + + mock_consumer_instance = AsyncMock() + mock_consumer_instance.__aiter__ = mock_aiter + mock_consumer_class.return_value = mock_consumer_instance + + await consumer.start() + + # Consume one message + task = consumer.consume_messages() + try: + await task + except StopAsyncIteration: + pass + + # Verify the moderation service was called with correct parameters + mock_moderation_service.process_moderation_response.assert_called_once_with( + object_id=123, approved=True + ) + + @pytest.mark.asyncio + async def test_consume_messages_handles_deserialization_failure( + self, consumer: KafkaConsumer, mock_moderation_service: AsyncMock + ) -> None: + """Test that consume_messages handles deserialization failures gracefully.""" + with patch( + "modservice.kafka.consumer.AIOKafkaConsumer" + ) as mock_consumer_class: + # Create an invalid message + mock_message = MagicMock() + mock_message.value = b"invalid protobuf data" + mock_message.key = b"\x00\x00\x00\x00\x00\x00\x00\x7b" + + # Mock the consumer to yield one message then stop + async def mock_aiter(self): # noqa: RUF029 + yield mock_message + + mock_consumer_instance = AsyncMock() + mock_consumer_instance.__aiter__ = mock_aiter + mock_consumer_class.return_value = mock_consumer_instance + + await consumer.start() + + # Consume one message - should not raise exception + task = consumer.consume_messages() + try: + await task + except StopAsyncIteration: + pass + + # Verify the moderation service was NOT called + mock_moderation_service.process_moderation_response.assert_not_called() + + @pytest.mark.asyncio + async def test_consume_messages_extracts_object_id_from_key( + self, consumer: KafkaConsumer, mock_moderation_service: AsyncMock + ) -> None: + """Test that consume_messages correctly extracts object_id from message key.""" + with patch( + "modservice.kafka.consumer.AIOKafkaConsumer" + ) as mock_consumer_class: + # Create a valid moderation response + response = ModerateObjectResponse(success=False) + message_bytes = response.SerializeToString() + + # Create message key with specific object_id + object_id = 456 + object_id_bytes = object_id.to_bytes(8, byteorder="big") + + mock_message = MagicMock() + mock_message.value = message_bytes + mock_message.key = object_id_bytes + + # Mock the consumer + async def mock_aiter(self): # noqa: RUF029 + yield mock_message + + mock_consumer_instance = AsyncMock() + mock_consumer_instance.__aiter__ = mock_aiter + mock_consumer_class.return_value = mock_consumer_instance + + await consumer.start() + + # Consume one message + task = consumer.consume_messages() + try: + await task + except StopAsyncIteration: + pass + + # Verify the correct object_id was extracted + mock_moderation_service.process_moderation_response.assert_called_once_with( + object_id=456, approved=False + ) + + @pytest.mark.asyncio + async def test_consume_messages_handles_processing_exception( + self, consumer: KafkaConsumer, mock_moderation_service: AsyncMock + ) -> None: + """Test that consume_messages handles processing exceptions gracefully.""" + with patch( + "modservice.kafka.consumer.AIOKafkaConsumer" + ) as mock_consumer_class: + # Create a valid moderation response + response = ModerateObjectResponse(success=True) + message_bytes = response.SerializeToString() + object_id_bytes = (789).to_bytes(8, byteorder="big") + + mock_message = MagicMock() + mock_message.value = message_bytes + mock_message.key = object_id_bytes + + # Make the moderation service raise an exception + mock_moderation_service.process_moderation_response.side_effect = ( + Exception("Processing error") + ) + + # Mock the consumer + async def mock_aiter(self): # noqa: RUF029 + yield mock_message + + mock_consumer_instance = AsyncMock() + mock_consumer_instance.__aiter__ = mock_aiter + mock_consumer_class.return_value = mock_consumer_instance + + await consumer.start() + + # Consume one message - should not raise exception + task = consumer.consume_messages() + try: + await task + except StopAsyncIteration: + pass + + # Verify the moderation service was called + mock_moderation_service.process_moderation_response.assert_called_once() diff --git a/tests/kafka/test_producer.py b/tests/kafka/test_producer.py new file mode 100644 index 0000000..87147ea --- /dev/null +++ b/tests/kafka/test_producer.py @@ -0,0 +1,174 @@ +"""Unit tests for Kafka producer.""" + +from unittest.mock import AsyncMock, patch + +import pytest + +from modservice.grpc.moderation_pb2 import ModerateObjectRequest, ObjectType +from modservice.kafka.producer import KafkaProducer + + +class TestKafkaProducer: + """Test suite for KafkaProducer class.""" + + @pytest.fixture + def producer(self) -> KafkaProducer: + """Create a KafkaProducer instance for testing.""" + return KafkaProducer( + bootstrap_servers="localhost:9092", + request_topic="moderation.requests", + ) + + @pytest.mark.asyncio + async def test_start_initializes_producer( + self, producer: KafkaProducer + ) -> None: + """Test that start() initializes the AIOKafkaProducer connection.""" + with patch( + "modservice.kafka.producer.AIOKafkaProducer" + ) as mock_producer_class: + mock_producer_instance = AsyncMock() + mock_producer_class.return_value = mock_producer_instance + + await producer.start() + + mock_producer_class.assert_called_once_with( + bootstrap_servers="localhost:9092" + ) + mock_producer_instance.start.assert_called_once() + + @pytest.mark.asyncio + async def test_stop_closes_producer(self, producer: KafkaProducer) -> None: + """Test that stop() closes the producer connection.""" + with patch( + "modservice.kafka.producer.AIOKafkaProducer" + ) as mock_producer_class: + mock_producer_instance = AsyncMock() + mock_producer_class.return_value = mock_producer_instance + + await producer.start() + await producer.stop() + + mock_producer_instance.stop.assert_called_once() + + @pytest.mark.asyncio + async def test_stop_without_start_does_nothing( + self, producer: KafkaProducer + ) -> None: + """Test that stop() does nothing if producer was never started.""" + # Should not raise an exception + await producer.stop() + + @pytest.mark.asyncio + async def test_send_moderation_request_serializes_and_sends( + self, producer: KafkaProducer + ) -> None: + """Test that send_moderation_request serializes and sends the message.""" + with patch( + "modservice.kafka.producer.AIOKafkaProducer" + ) as mock_producer_class: + mock_producer_instance = AsyncMock() + mock_producer_class.return_value = mock_producer_instance + + await producer.start() + + # Send a moderation request + await producer.send_moderation_request( + object_id=123, + object_type=ObjectType.OBJECT_TYPE_MOD_DESCRIPTION, + text="Test mod title Test mod description", + ) + + # Verify send was called + mock_producer_instance.send.assert_called_once() + call_args = mock_producer_instance.send.call_args + + # Verify topic (first positional argument) + assert call_args[0][0] == "moderation.requests" + + # Verify message can be deserialized (keyword argument 'value') + message_bytes = call_args[1]["value"] + request = ModerateObjectRequest() + request.ParseFromString(message_bytes) + + assert request.id == 123 + assert request.type == ObjectType.OBJECT_TYPE_MOD_DESCRIPTION + assert request.text == "Test mod title Test mod description" + + # Verify key is encoded correctly (keyword argument 'key') + key_bytes = call_args[1]["key"] + object_id = int.from_bytes(key_bytes, byteorder="big") + assert object_id == 123 + + @pytest.mark.asyncio + async def test_send_moderation_request_without_start_raises_error( + self, producer: KafkaProducer + ) -> None: + """Test that send_moderation_request raises error if producer not started.""" + with pytest.raises(RuntimeError, match="Producer not started"): + await producer.send_moderation_request( + object_id=123, + object_type=ObjectType.OBJECT_TYPE_MOD_DESCRIPTION, + text="Test text", + ) + + @pytest.mark.asyncio + async def test_send_moderation_request_with_empty_text( + self, producer: KafkaProducer + ) -> None: + """Test that send_moderation_request handles empty text.""" + with patch( + "modservice.kafka.producer.AIOKafkaProducer" + ) as mock_producer_class: + mock_producer_instance = AsyncMock() + mock_producer_class.return_value = mock_producer_instance + + await producer.start() + + # Send with empty text + await producer.send_moderation_request( + object_id=456, + object_type=ObjectType.OBJECT_TYPE_MOD_DESCRIPTION, + text="", + ) + + # Verify send was called + mock_producer_instance.send.assert_called_once() + call_args = mock_producer_instance.send.call_args + + # Verify message (keyword argument 'value') + message_bytes = call_args[1]["value"] + request = ModerateObjectRequest() + request.ParseFromString(message_bytes) + + assert request.id == 456 + assert request.text == "" + + @pytest.mark.asyncio + async def test_send_moderation_request_with_special_characters( + self, producer: KafkaProducer + ) -> None: + """Test that send_moderation_request handles special characters in text.""" + with patch( + "modservice.kafka.producer.AIOKafkaProducer" + ) as mock_producer_class: + mock_producer_instance = AsyncMock() + mock_producer_class.return_value = mock_producer_instance + + await producer.start() + + special_text = "Test with émojis 🎮 and symbols !@#$%" + await producer.send_moderation_request( + object_id=789, + object_type=ObjectType.OBJECT_TYPE_MOD_DESCRIPTION, + text=special_text, + ) + + # Verify message + call_args = mock_producer_instance.send.call_args + message_bytes = call_args[1]["value"] + request = ModerateObjectRequest() + request.ParseFromString(message_bytes) + + assert request.id == 789 + assert request.text == special_text diff --git a/tests/kafka/test_property_response_extraction.py b/tests/kafka/test_property_response_extraction.py new file mode 100644 index 0000000..d3c93e8 --- /dev/null +++ b/tests/kafka/test_property_response_extraction.py @@ -0,0 +1,86 @@ +"""Property-based tests for moderation response extraction. + +**Feature: mod-kafka-moderation** +""" + +from unittest.mock import AsyncMock, MagicMock + +import pytest +from hypothesis import given, settings +from hypothesis import strategies as st + +from modservice.grpc.moderation_pb2 import ModerateObjectResponse + + +class TestModerationResponseExtractionProperty: + """Property-based tests for moderation response extraction. + + **Validates: Requirements 5.5** + """ + + @given( + object_id=st.integers(min_value=1, max_value=2**63 - 1), + success=st.booleans(), + ) + @settings(max_examples=100) + @pytest.mark.asyncio + async def test_property_5_moderation_response_extraction( + self, + object_id: int, + success: bool, + ) -> None: + """Property 5: Moderation response extraction. + + **Feature: mod-kafka-moderation, Property 5: For any + ModerateObjectResponse message with an object_id, the consumer should + correctly extract both the object_id and the success field value** + + For any ModerateObjectResponse message with an object_id, the consumer + should correctly extract both the object_id and the success field value. + """ + # Create a mock moderation service to capture the extracted values + mock_moderation_service = AsyncMock() + mock_moderation_service.process_moderation_response = AsyncMock() + + # Create a ModerateObjectResponse message with the generated values + response = ModerateObjectResponse(success=success) + message_bytes = response.SerializeToString() + + # Encode object_id as bytes (big endian, 8 bytes for int64) + object_id_bytes = object_id.to_bytes(8, byteorder="big") + + # Create a mock Kafka message + mock_message = MagicMock() + mock_message.value = message_bytes + mock_message.key = object_id_bytes + + # Test the extraction logic directly (this is what the consumer does) + # Deserialize the protobuf message + response_parsed = ModerateObjectResponse() + response_parsed.ParseFromString(mock_message.value) + + # Extract object_id from message key (big endian encoding) + extracted_object_id = int.from_bytes(mock_message.key, byteorder="big") + + # Simulate calling the moderation service + await mock_moderation_service.process_moderation_response( + object_id=extracted_object_id, approved=response_parsed.success + ) + + # Assert that the extraction correctly preserved both fields + mock_moderation_service.process_moderation_response.assert_called_once_with( + object_id=object_id, approved=success + ) + + # Verify the extracted values match the original values + call_args = ( + mock_moderation_service.process_moderation_response.call_args + ) + assert call_args.kwargs["object_id"] == object_id, ( + f"Extracted object_id {call_args.kwargs['object_id']} should match " + f"original {object_id}" + ) + assert call_args.kwargs["approved"] == success, ( + f"Extracted success {call_args.kwargs['approved']} should match " + f"original {success}" + ) diff --git a/tests/moderation/__init__.py b/tests/moderation/__init__.py new file mode 100644 index 0000000..7a1620a --- /dev/null +++ b/tests/moderation/__init__.py @@ -0,0 +1 @@ +"""Tests for moderation service.""" diff --git a/tests/moderation/test_integration_approved_flow.py b/tests/moderation/test_integration_approved_flow.py new file mode 100644 index 0000000..0f141f0 --- /dev/null +++ b/tests/moderation/test_integration_approved_flow.py @@ -0,0 +1,303 @@ +"""Integration test for approved moderation flow. + +Tests the complete flow: +1. Create mod with ON_MODERATION status +2. Send moderation request to Kafka +3. Receive approved response from Kafka +4. Status updates to UPLOADED + +Validates Requirements: 2.1, 5.1, 5.2, 5.3, 5.4, 6.1, 6.2, 6.6 +""" + +import asyncio +from unittest.mock import AsyncMock, Mock, patch, call + +import pytest + +from modservice.grpc import moderation_pb2 +from modservice.moderation.moderation_service import ModerationService +from modservice.repository.repository import ModRepository + + +@pytest.fixture +def mock_db_pool(): + """Create a mock database pool.""" + pool = Mock() + conn = Mock() + + # Mock connection context manager + conn.__aenter__ = AsyncMock(return_value=conn) + conn.__aexit__ = AsyncMock(return_value=None) + + # Mock fetchval for create_mod + conn.fetchval = AsyncMock(return_value=123) + + # Mock execute for set_status + conn.execute = AsyncMock(return_value="UPDATE 1") + + pool.acquire = Mock(return_value=conn) + + return pool + + +@pytest.fixture +def repository(mock_db_pool): + """Create a ModRepository with mocked database.""" + return ModRepository(mock_db_pool) + + +@pytest.fixture +def mock_kafka_producer(): + """Create a mock Kafka producer.""" + with patch('modservice.moderation.moderation_service.KafkaProducer') as mock_producer_class: + producer_instance = Mock() + producer_instance.send_moderation_request = Mock(return_value=True) + producer_instance.flush = Mock() + producer_instance.topic = "moderation-request" + mock_producer_class.return_value = producer_instance + yield producer_instance + + +@pytest.fixture +def mock_kafka_consumer(): + """Create a mock Kafka consumer.""" + with patch('modservice.moderation.moderation_service.KafkaConsumer') as mock_consumer_class: + consumer_instance = Mock() + consumer_instance.start = Mock() + consumer_instance.stop = Mock() + mock_consumer_class.return_value = consumer_instance + yield consumer_instance + + +@pytest.fixture +def moderation_service(repository, mock_kafka_producer, mock_kafka_consumer): + """Create a ModerationService with mocked Kafka components.""" + # Create a new event loop for the service + loop = asyncio.new_event_loop() + service = ModerationService(repository, loop) + + # Start the loop in a thread to handle async callbacks + import threading + def run_loop(): + asyncio.set_event_loop(loop) + loop.run_forever() + + thread = threading.Thread(target=run_loop, daemon=True) + thread.start() + + yield service + + # Cleanup + loop.call_soon_threadsafe(loop.stop) + thread.join(timeout=1) + loop.close() + + +@pytest.mark.asyncio +async def test_approved_mod_flow_integration( + repository: ModRepository, + moderation_service: ModerationService, + mock_kafka_producer: Mock, + caplog: pytest.LogCaptureFixture +) -> None: + """Test complete approved moderation flow. + + Validates Requirements: + - 2.1: Mod created with ON_MODERATION status + - 5.1: Serializes to ModerateObjectRequest protobuf + - 5.2: Sets type to OBJECT_TYPE_MOD_DESCRIPTION + - 5.3: Uses mod_id as Kafka message key + - 5.4: Produces to moderation-request topic + - 6.1: Extracts mod_id from message key + - 6.2: Updates status to UPLOADED on success=true + - 6.6: Completes processing without exceptions + """ + import logging + caplog.set_level(logging.INFO) + + # Step 1: Create mod with ON_MODERATION status + mod_id = await repository.create_mod( + title="Test Mod", + author_id=1, + description="Test Description" + ) + + assert mod_id == 123 + + # Step 2: Send moderation request to Kafka + success = moderation_service.send_moderation_request( + mod_id=mod_id, + title="Test Mod", + description="Test Description" + ) + + assert success is True + + # Verify Kafka producer was called with correct protobuf message + mock_kafka_producer.send_moderation_request.assert_called_once() + call_args = mock_kafka_producer.send_moderation_request.call_args + request = call_args[0][0] + + # Validate Requirements 5.1, 5.2, 5.3 + assert isinstance(request, moderation_pb2.ModerateObjectRequest) + assert request.id == mod_id + assert request.text == "Test Mod Test Description" + assert request.type == moderation_pb2.OBJECT_TYPE_MOD_DESCRIPTION + + # Step 3: Simulate receiving approved response from Kafka + response = moderation_pb2.ModerateObjectResponse() + response.success = True + + # Call the callback directly to simulate Kafka consumer receiving message + moderation_service._handle_moderation_response(response, mod_id) + + # Wait for async status update to complete + await asyncio.sleep(0.2) + + # Step 4: Verify status was updated to UPLOADED + # Check that set_status was called with correct parameters + repository._db_pool.acquire.return_value.execute.assert_called() + + # Verify the correct status was set + conn = repository._db_pool.acquire.return_value + call_args = conn.execute.call_args + assert call_args[0][1] == "UPLOADED" # First parameter is status + assert call_args[0][2] == mod_id # Second parameter is mod_id + + +@pytest.mark.asyncio +async def test_approved_flow_with_actual_status_update( + repository: ModRepository, + moderation_service: ModerationService +) -> None: + """Test that status update is actually called with correct parameters. + + Validates Requirements 6.2, 6.6: + - Status updates to UPLOADED when approved=true + - Processing completes without exceptions + """ + mod_id = 456 + + # Simulate approved response + response = moderation_pb2.ModerateObjectResponse() + response.success = True + + # Call the callback + moderation_service._handle_moderation_response(response, mod_id) + + # Wait for async operation + await asyncio.sleep(0.2) + + # Verify set_status was called with UPLOADED + conn = repository._db_pool.acquire.return_value + conn.execute.assert_called_with( + """ + UPDATE mods + SET status = $1 + WHERE id = $2 + """, + "UPLOADED", + mod_id + ) + + +@pytest.mark.asyncio +async def test_kafka_message_key_matches_mod_id( + moderation_service: ModerationService, + mock_kafka_producer: Mock +) -> None: + """Test that Kafka message key equals mod_id. + + Validates Requirement 5.3: + - Kafka message key must equal mod_id + """ + mod_id = 789 + + # Send moderation request + moderation_service.send_moderation_request( + mod_id=mod_id, + title="Test", + description="Test" + ) + + # Verify the request was created with correct mod_id + call_args = mock_kafka_producer.send_moderation_request.call_args + request = call_args[0][0] + + assert request.id == mod_id + + +@pytest.mark.asyncio +async def test_moderation_request_sent_to_correct_topic( + moderation_service: ModerationService, + mock_kafka_producer: Mock +) -> None: + """Test that moderation request is sent to moderation-request topic. + + Validates Requirement 5.4: + - Message must be produced to moderation-request topic + """ + # Verify producer was initialized with correct topic + assert moderation_service.producer.topic == "moderation-request" + + # Send request + moderation_service.send_moderation_request( + mod_id=999, + title="Test", + description="Test" + ) + + # Verify send was called + mock_kafka_producer.send_moderation_request.assert_called_once() + + +@pytest.mark.asyncio +async def test_processing_completes_without_exceptions( + moderation_service: ModerationService, + repository: ModRepository +) -> None: + """Test that successful processing completes without exceptions. + + Validates Requirement 6.6: + - When status update succeeds, processing must complete without exceptions + """ + mod_id = 111 + + # Simulate approved response + response = moderation_pb2.ModerateObjectResponse() + response.success = True + + # This should not raise any exceptions + try: + moderation_service._handle_moderation_response(response, mod_id) + await asyncio.sleep(0.2) + except Exception as e: + pytest.fail(f"Processing raised unexpected exception: {e}") + + +@pytest.mark.asyncio +async def test_mod_id_extraction_from_response( + moderation_service: ModerationService, + repository: ModRepository +) -> None: + """Test that mod_id is correctly extracted from Kafka message. + + Validates Requirement 6.1: + - ModerationService SHALL extract mod_id from message key + """ + mod_id = 222 + + # Simulate response with specific mod_id + response = moderation_pb2.ModerateObjectResponse() + response.success = True + + # Call handler with mod_id + moderation_service._handle_moderation_response(response, mod_id) + + # Wait for async operation + await asyncio.sleep(0.2) + + # Verify the correct mod_id was used in status update + conn = repository._db_pool.acquire.return_value + call_args = conn.execute.call_args + assert call_args[0][2] == mod_id # Second parameter is mod_id diff --git a/tests/moderation/test_integration_kafka_failure.py b/tests/moderation/test_integration_kafka_failure.py new file mode 100644 index 0000000..da606c9 --- /dev/null +++ b/tests/moderation/test_integration_kafka_failure.py @@ -0,0 +1,323 @@ +"""Integration test for Kafka failure handling. + +Tests the complete flow when Kafka fails: +1. Create mod with ON_MODERATION status +2. Kafka send fails (buffer full, connection error, etc.) +3. Status remains ON_MODERATION + +Validates Requirements: 7.1, 7.2, 7.3, 7.4 +""" + +import asyncio +from unittest.mock import AsyncMock, Mock, patch + +import pytest + +from modservice.grpc import moderation_pb2 +from modservice.moderation.moderation_service import ModerationService +from modservice.repository.repository import ModRepository + + +@pytest.fixture +def mock_db_pool(): + """Create a mock database pool.""" + pool = Mock() + conn = Mock() + + # Mock connection context manager + conn.__aenter__ = AsyncMock(return_value=conn) + conn.__aexit__ = AsyncMock(return_value=None) + + # Mock fetchval for create_mod - returns mod_id + conn.fetchval = AsyncMock(return_value=999) + + # Mock execute for set_status + conn.execute = AsyncMock(return_value="UPDATE 1") + + pool.acquire = Mock(return_value=conn) + + return pool + + +@pytest.fixture +def repository(mock_db_pool): + """Create a ModRepository with mocked database.""" + return ModRepository(mock_db_pool) + + +@pytest.fixture +def mock_kafka_producer_failure(): + """Create a mock Kafka producer that fails to send.""" + with patch('modservice.moderation.moderation_service.KafkaProducer') as mock_producer_class: + producer_instance = Mock() + # Simulate Kafka send failure + producer_instance.send_moderation_request = Mock(return_value=False) + producer_instance.flush = Mock() + producer_instance.topic = "moderation-request" + mock_producer_class.return_value = producer_instance + yield producer_instance + + +@pytest.fixture +def mock_kafka_consumer(): + """Create a mock Kafka consumer.""" + with patch('modservice.moderation.moderation_service.KafkaConsumer') as mock_consumer_class: + consumer_instance = Mock() + consumer_instance.start = Mock() + consumer_instance.stop = Mock() + mock_consumer_class.return_value = consumer_instance + yield consumer_instance + + +@pytest.fixture +def moderation_service_with_failing_kafka(repository, mock_kafka_producer_failure, mock_kafka_consumer): + """Create a ModerationService with failing Kafka producer.""" + # Create a new event loop for the service + loop = asyncio.new_event_loop() + service = ModerationService(repository, loop) + + # Start the loop in a thread to handle async callbacks + import threading + def run_loop(): + asyncio.set_event_loop(loop) + loop.run_forever() + + thread = threading.Thread(target=run_loop, daemon=True) + thread.start() + + yield service + + # Cleanup + loop.call_soon_threadsafe(loop.stop) + thread.join(timeout=1) + loop.close() + + +@pytest.mark.asyncio +async def test_kafka_failure_mod_remains_on_moderation( + repository: ModRepository, + moderation_service_with_failing_kafka: ModerationService, + mock_kafka_producer_failure: Mock, + caplog: pytest.LogCaptureFixture +) -> None: + """Test that mod remains in ON_MODERATION status when Kafka send fails. + + Validates Requirements: + - 7.1: KafkaProducer retries up to 3 times on buffer full error + - 7.2: KafkaProducer retries up to 3 times on connection error + - 7.3: ModerationService returns false and logs error when all retries fail + - 7.4: Mod remains in ON_MODERATION status when send_moderation_request returns false + """ + import logging + caplog.set_level(logging.INFO) + + # Step 1: Create mod with ON_MODERATION status + mod_id = await repository.create_mod( + title="Test Mod", + author_id=1, + description="Test Description" + ) + + assert mod_id == 999 + + # Verify mod was created (fetchval was called) + conn = repository._db_pool.acquire.return_value + conn.fetchval.assert_called_once() + + # Step 2: Attempt to send moderation request to Kafka (will fail) + success = moderation_service_with_failing_kafka.send_moderation_request( + mod_id=mod_id, + title="Test Mod", + description="Test Description" + ) + + # Verify send_moderation_request returns False (Requirement 7.3) + assert success is False + + # Verify Kafka producer was called but failed + mock_kafka_producer_failure.send_moderation_request.assert_called_once() + + # Verify error was logged (Requirement 7.3) + assert any("Failed to send moderation request" in record.message for record in caplog.records) + assert any(f"mod_id={mod_id}" in record.message for record in caplog.records) + + # Step 3: Verify status was NOT updated (remains ON_MODERATION) + # The set_status method should NOT have been called since Kafka send failed + # and no moderation response was received + + # Wait a bit to ensure no async status update happens + await asyncio.sleep(0.2) + + # Verify set_status was NOT called (mod remains in ON_MODERATION) + # The execute method should only have been called once for create_mod, not for set_status + # Actually, create_mod uses fetchval, not execute, so execute should not be called at all + conn.execute.assert_not_called() + + +@pytest.mark.asyncio +async def test_kafka_buffer_full_error_handling( + moderation_service_with_failing_kafka: ModerationService, + mock_kafka_producer_failure: Mock, + caplog: pytest.LogCaptureFixture +) -> None: + """Test handling of Kafka buffer full error. + + Validates Requirements: + - 7.1: KafkaProducer SHALL retry up to 3 times on buffer full error + - 7.3: ModerationService SHALL return false and log error when all retries fail + """ + import logging + caplog.set_level(logging.ERROR) + + mod_id = 123 + + # Attempt to send (will fail due to buffer full) + success = moderation_service_with_failing_kafka.send_moderation_request( + mod_id=mod_id, + title="Test", + description="Test" + ) + + # Verify failure + assert success is False + + # Verify error was logged + assert any("Failed to send moderation request" in record.message for record in caplog.records) + + +@pytest.mark.asyncio +async def test_kafka_connection_error_handling( + moderation_service_with_failing_kafka: ModerationService, + mock_kafka_producer_failure: Mock, + caplog: pytest.LogCaptureFixture +) -> None: + """Test handling of Kafka connection error. + + Validates Requirements: + - 7.2: KafkaProducer SHALL retry up to 3 times on connection error + - 7.3: ModerationService SHALL return false and log error when all retries fail + """ + import logging + caplog.set_level(logging.ERROR) + + mod_id = 456 + + # Attempt to send (will fail due to connection error) + success = moderation_service_with_failing_kafka.send_moderation_request( + mod_id=mod_id, + title="Test", + description="Test" + ) + + # Verify failure + assert success is False + + # Verify error was logged + assert any("Failed to send moderation request" in record.message for record in caplog.records) + + +@pytest.mark.asyncio +async def test_mod_creation_continues_despite_kafka_failure( + repository: ModRepository, + moderation_service_with_failing_kafka: ModerationService +) -> None: + """Test that mod creation continues even if Kafka send fails. + + Validates Requirement 7.5: + - ModService SHALL continue mod creation even if Kafka send fails + """ + # Create mod + mod_id = await repository.create_mod( + title="Test Mod", + author_id=1, + description="Test Description" + ) + + # Verify mod was created successfully + assert mod_id == 999 + + # Attempt to send to Kafka (will fail) + success = moderation_service_with_failing_kafka.send_moderation_request( + mod_id=mod_id, + title="Test Mod", + description="Test Description" + ) + + # Kafka send fails, but mod creation was successful + assert success is False + assert mod_id is not None + + # The mod exists in the database with ON_MODERATION status + # (it was not rolled back due to Kafka failure) + + +@pytest.mark.asyncio +async def test_multiple_kafka_failures_do_not_update_status( + repository: ModRepository, + moderation_service_with_failing_kafka: ModerationService +) -> None: + """Test that multiple Kafka failures do not trigger status updates. + + Validates Requirement 7.4: + - Mod SHALL remain in ON_MODERATION status when send_moderation_request returns false + """ + mod_ids = [111, 222, 333] + + for mod_id in mod_ids: + # Mock create_mod to return different mod_ids + conn = repository._db_pool.acquire.return_value + conn.fetchval = AsyncMock(return_value=mod_id) + + # Create mod + created_id = await repository.create_mod( + title=f"Mod {mod_id}", + author_id=1, + description="Test" + ) + + assert created_id == mod_id + + # Attempt to send to Kafka (will fail) + success = moderation_service_with_failing_kafka.send_moderation_request( + mod_id=mod_id, + title=f"Mod {mod_id}", + description="Test" + ) + + assert success is False + + # Wait to ensure no async status updates happen + await asyncio.sleep(0.3) + + # Verify set_status was never called for any mod + conn = repository._db_pool.acquire.return_value + conn.execute.assert_not_called() + + +@pytest.mark.asyncio +async def test_kafka_failure_logs_mod_id( + moderation_service_with_failing_kafka: ModerationService, + caplog: pytest.LogCaptureFixture +) -> None: + """Test that Kafka failure logs include mod_id for debugging. + + Validates Requirement 5.7: + - When KafkaProducer fails, ModerationService SHALL log error with mod_id + """ + import logging + caplog.set_level(logging.ERROR) + + mod_id = 789 + + # Attempt to send (will fail) + success = moderation_service_with_failing_kafka.send_moderation_request( + mod_id=mod_id, + title="Test", + description="Test" + ) + + assert success is False + + # Verify mod_id is in error log + error_logs = [record.message for record in caplog.records if record.levelname == "ERROR"] + assert any(str(mod_id) in log for log in error_logs) diff --git a/tests/moderation/test_integration_rejected_flow.py b/tests/moderation/test_integration_rejected_flow.py new file mode 100644 index 0000000..6ffe7b4 --- /dev/null +++ b/tests/moderation/test_integration_rejected_flow.py @@ -0,0 +1,282 @@ +"""Integration test for rejected moderation flow. + +Tests the complete flow: +1. Create mod with ON_MODERATION status +2. Send moderation request to Kafka +3. Receive rejected response from Kafka +4. Status updates to HIDDEN + +Validates Requirements: 2.1, 6.1, 6.3, 6.6 +""" + +import asyncio +from unittest.mock import AsyncMock, Mock, patch + +import pytest + +from modservice.grpc import moderation_pb2 +from modservice.moderation.moderation_service import ModerationService +from modservice.repository.repository import ModRepository + + +@pytest.fixture +def mock_db_pool(): + """Create a mock database pool.""" + pool = Mock() + conn = Mock() + + # Mock connection context manager + conn.__aenter__ = AsyncMock(return_value=conn) + conn.__aexit__ = AsyncMock(return_value=None) + + # Mock fetchval for create_mod + conn.fetchval = AsyncMock(return_value=456) + + # Mock execute for set_status + conn.execute = AsyncMock(return_value="UPDATE 1") + + pool.acquire = Mock(return_value=conn) + + return pool + + +@pytest.fixture +def repository(mock_db_pool): + """Create a ModRepository with mocked database.""" + return ModRepository(mock_db_pool) + + +@pytest.fixture +def mock_kafka_producer(): + """Create a mock Kafka producer.""" + with patch('modservice.moderation.moderation_service.KafkaProducer') as mock_producer_class: + producer_instance = Mock() + producer_instance.send_moderation_request = Mock(return_value=True) + producer_instance.flush = Mock() + producer_instance.topic = "moderation-request" + mock_producer_class.return_value = producer_instance + yield producer_instance + + +@pytest.fixture +def mock_kafka_consumer(): + """Create a mock Kafka consumer.""" + with patch('modservice.moderation.moderation_service.KafkaConsumer') as mock_consumer_class: + consumer_instance = Mock() + consumer_instance.start = Mock() + consumer_instance.stop = Mock() + mock_consumer_class.return_value = consumer_instance + yield consumer_instance + + +@pytest.fixture +def moderation_service(repository, mock_kafka_producer, mock_kafka_consumer): + """Create a ModerationService with mocked Kafka components.""" + # Create a new event loop for the service + loop = asyncio.new_event_loop() + service = ModerationService(repository, loop) + + # Start the loop in a thread to handle async callbacks + import threading + def run_loop(): + asyncio.set_event_loop(loop) + loop.run_forever() + + thread = threading.Thread(target=run_loop, daemon=True) + thread.start() + + yield service + + # Cleanup + loop.call_soon_threadsafe(loop.stop) + thread.join(timeout=1) + loop.close() + + +@pytest.mark.asyncio +async def test_rejected_mod_flow_integration( + repository: ModRepository, + moderation_service: ModerationService, + mock_kafka_producer: Mock, + caplog: pytest.LogCaptureFixture +) -> None: + """Test complete rejected moderation flow. + + Validates Requirements: + - 2.1: Mod created with ON_MODERATION status + - 6.1: Extracts mod_id from message key + - 6.3: Updates status to HIDDEN on success=false + - 6.6: Completes processing without exceptions + """ + import logging + caplog.set_level(logging.INFO) + + # Step 1: Create mod with ON_MODERATION status + mod_id = await repository.create_mod( + title="Rejected Mod", + author_id=2, + description="This mod will be rejected" + ) + + assert mod_id == 456 + + # Step 2: Send moderation request to Kafka + success = moderation_service.send_moderation_request( + mod_id=mod_id, + title="Rejected Mod", + description="This mod will be rejected" + ) + + assert success is True + + # Verify Kafka producer was called + mock_kafka_producer.send_moderation_request.assert_called_once() + call_args = mock_kafka_producer.send_moderation_request.call_args + request = call_args[0][0] + + # Validate request structure + assert isinstance(request, moderation_pb2.ModerateObjectRequest) + assert request.id == mod_id + assert request.type == moderation_pb2.OBJECT_TYPE_MOD_DESCRIPTION + + # Step 3: Simulate receiving rejected response from Kafka + response = moderation_pb2.ModerateObjectResponse() + response.success = False + + # Call the callback directly to simulate Kafka consumer receiving message + moderation_service._handle_moderation_response(response, mod_id) + + # Wait for async status update to complete + await asyncio.sleep(0.2) + + # Step 4: Verify status was updated to HIDDEN + # Check that set_status was called with correct parameters + repository._db_pool.acquire.return_value.execute.assert_called() + + # Verify the correct status was set + conn = repository._db_pool.acquire.return_value + call_args = conn.execute.call_args + assert call_args[0][1] == "HIDDEN" # First parameter is status + assert call_args[0][2] == mod_id # Second parameter is mod_id + + +@pytest.mark.asyncio +async def test_rejected_flow_with_actual_status_update( + repository: ModRepository, + moderation_service: ModerationService +) -> None: + """Test that status update is actually called with HIDDEN for rejected mods. + + Validates Requirements 6.3, 6.6: + - Status updates to HIDDEN when success=false + - Processing completes without exceptions + """ + mod_id = 789 + + # Simulate rejected response + response = moderation_pb2.ModerateObjectResponse() + response.success = False + + # Call the callback + moderation_service._handle_moderation_response(response, mod_id) + + # Wait for async operation + await asyncio.sleep(0.2) + + # Verify set_status was called with HIDDEN + conn = repository._db_pool.acquire.return_value + conn.execute.assert_called_with( + """ + UPDATE mods + SET status = $1 + WHERE id = $2 + """, + "HIDDEN", + mod_id + ) + + +@pytest.mark.asyncio +async def test_rejected_processing_completes_without_exceptions( + moderation_service: ModerationService, + repository: ModRepository +) -> None: + """Test that rejected mod processing completes without exceptions. + + Validates Requirement 6.6: + - When status update succeeds, processing must complete without exceptions + """ + mod_id = 333 + + # Simulate rejected response + response = moderation_pb2.ModerateObjectResponse() + response.success = False + + # This should not raise any exceptions + try: + moderation_service._handle_moderation_response(response, mod_id) + await asyncio.sleep(0.2) + except Exception as e: + pytest.fail(f"Processing raised unexpected exception: {e}") + + +@pytest.mark.asyncio +async def test_mod_id_extraction_from_rejected_response( + moderation_service: ModerationService, + repository: ModRepository +) -> None: + """Test that mod_id is correctly extracted from rejected Kafka message. + + Validates Requirement 6.1: + - ModerationService SHALL extract mod_id from message key + """ + mod_id = 555 + + # Simulate rejected response with specific mod_id + response = moderation_pb2.ModerateObjectResponse() + response.success = False + + # Call handler with mod_id + moderation_service._handle_moderation_response(response, mod_id) + + # Wait for async operation + await asyncio.sleep(0.2) + + # Verify the correct mod_id was used in status update + conn = repository._db_pool.acquire.return_value + call_args = conn.execute.call_args + assert call_args[0][2] == mod_id # Second parameter is mod_id + + +@pytest.mark.asyncio +async def test_multiple_rejected_mods_processed_correctly( + moderation_service: ModerationService, + repository: ModRepository +) -> None: + """Test that multiple rejected mods are processed correctly. + + Validates Requirements 6.1, 6.3, 6.6: + - Each mod_id is extracted correctly + - Each mod status is updated to HIDDEN + - All processing completes without exceptions + """ + mod_ids = [111, 222, 333] + + for mod_id in mod_ids: + # Simulate rejected response + response = moderation_pb2.ModerateObjectResponse() + response.success = False + + # Process each rejection + moderation_service._handle_moderation_response(response, mod_id) + + # Wait for all async operations + await asyncio.sleep(0.3) + + # Verify all mods were updated to HIDDEN + conn = repository._db_pool.acquire.return_value + assert conn.execute.call_count == len(mod_ids) + + # Check that all calls used HIDDEN status + for call in conn.execute.call_args_list: + assert call[0][1] == "HIDDEN" diff --git a/tests/moderation/test_moderation_service.py b/tests/moderation/test_moderation_service.py new file mode 100644 index 0000000..c4734a5 --- /dev/null +++ b/tests/moderation/test_moderation_service.py @@ -0,0 +1,119 @@ +"""Tests for ModerationService.""" + +from typing import Any +from unittest.mock import AsyncMock, Mock + +import pytest + +from modservice.grpc.moderation_pb2 import ObjectType +from modservice.moderation.moderation_service import ModerationService + + +@pytest.fixture +def mock_producer() -> Mock: + """Create a mock Kafka producer.""" + producer = Mock() + producer.send_moderation_request = AsyncMock() + return producer + + +@pytest.fixture +def mock_repository() -> Mock: + """Create a mock repository.""" + repository = Mock() + repository.set_status = AsyncMock() + return repository + + +@pytest.fixture +def moderation_service( + mock_producer: Mock, mock_repository: Mock +) -> ModerationService: + """Create a ModerationService instance with mocked dependencies.""" + return ModerationService(mock_producer, mock_repository) + + +@pytest.mark.asyncio +async def test_send_moderation_request_combines_title_and_description( + moderation_service: ModerationService, mock_producer: Mock +) -> None: + """Test that send_moderation_request combines title and description with single space.""" + mod_id = 123 + title = "Test Mod" + description = "This is a test mod" + + await moderation_service.send_moderation_request( + mod_id, title, description + ) + + mock_producer.send_moderation_request.assert_called_once_with( + object_id=mod_id, + object_type=ObjectType.OBJECT_TYPE_MOD_DESCRIPTION, + text="Test Mod This is a test mod", + ) + + +@pytest.mark.asyncio +async def test_send_moderation_request_handles_producer_failure( + moderation_service: ModerationService, mock_producer: Mock, caplog: Any +) -> None: + """Test that send_moderation_request logs errors without raising exceptions.""" + mock_producer.send_moderation_request.side_effect = Exception( + "Kafka error" + ) + + # Should not raise exception + await moderation_service.send_moderation_request( + 123, "Title", "Description" + ) + + # Should log error + assert "Failed to send moderation request for mod 123" in caplog.text + + +@pytest.mark.asyncio +async def test_process_moderation_response_approved( + moderation_service: ModerationService, mock_repository: Mock +) -> None: + """Test that approved response sets status to UPLOADED.""" + mock_repository.set_status.return_value = True + + await moderation_service.process_moderation_response(123, approved=True) + + mock_repository.set_status.assert_called_once_with(123, "UPLOADED") + + +@pytest.mark.asyncio +async def test_process_moderation_response_rejected( + moderation_service: ModerationService, mock_repository: Mock +) -> None: + """Test that rejected response sets status to BANNED.""" + mock_repository.set_status.return_value = True + + await moderation_service.process_moderation_response(123, approved=False) + + mock_repository.set_status.assert_called_once_with(123, "BANNED") + + +@pytest.mark.asyncio +async def test_process_moderation_response_mod_not_found( + moderation_service: ModerationService, mock_repository: Mock, caplog: Any +) -> None: + """Test that warning is logged when mod is not found.""" + mock_repository.set_status.return_value = False + + await moderation_service.process_moderation_response(123, approved=True) + + assert "Mod 123 not found for status update" in caplog.text + + +@pytest.mark.asyncio +async def test_process_moderation_response_handles_exception( + moderation_service: ModerationService, mock_repository: Mock, caplog: Any +) -> None: + """Test that exceptions during status update are logged.""" + mock_repository.set_status.side_effect = Exception("Database error") + + await moderation_service.process_moderation_response(123, approved=True) + + assert "Failed to update mod 123 status" in caplog.text diff --git a/tests/moderation/test_property_moderation_request_structure.py b/tests/moderation/test_property_moderation_request_structure.py new file mode 100644 index 0000000..d2e01a4 --- /dev/null +++ b/tests/moderation/test_property_moderation_request_structure.py @@ -0,0 +1,96 @@ +"""Property-based tests for moderation request message structure. + +**Feature: mod-kafka-moderation** +""" + +from unittest.mock import AsyncMock, Mock + +import pytest +from hypothesis import given, settings +from hypothesis import strategies as st + +from modservice.grpc.moderation_pb2 import ModerateObjectRequest, ObjectType +from modservice.moderation.moderation_service import ModerationService + + +class TestModerationRequestMessageStructureProperty: + """Property-based tests for moderation request message structure. + + **Validates: Requirements 4.2, 4.3, 4.4, 4.5, 6.3, 9.1** + """ + + @given( + mod_id=st.integers(min_value=1, max_value=1000000), + title=st.text(min_size=1, max_size=255), + description=st.text(min_size=1, max_size=1000), + ) + @settings(max_examples=100) + @pytest.mark.asyncio + async def test_property_3_moderation_request_message_structure( + self, + mod_id: int, + title: str, + description: str, + ) -> None: + """Property 3: Moderation request message structure. + + **Feature: mod-kafka-moderation, Property 3: For any mod with title + and description, when a moderation request is sent, the resulting + ModerateObjectRequest message should have correct id, type, and text + fields** + + For any mod with title and description, when a moderation request is sent, + the resulting ModerateObjectRequest message should have: + - id field equal to the mod_id + - type field equal to OBJECT_TYPE_MOD_DESCRIPTION + - text field equal to "{title} {description}" (title and description separated by exactly one space) + """ + # Track the message that was sent + captured_message = None + + async def capture_send_moderation_request( # noqa: RUF029 + object_id: int, object_type: ObjectType, text: str + ) -> None: + nonlocal captured_message + # Create the actual protobuf message that would be sent + captured_message = ModerateObjectRequest( + id=object_id, type=object_type, text=text + ) + + # Mock producer that captures the message + mock_producer = Mock() + mock_producer.send_moderation_request = AsyncMock( + side_effect=capture_send_moderation_request + ) + + # Mock repository (not used in this test) + mock_repository = Mock() + + # Create moderation service + moderation_service = ModerationService(mock_producer, mock_repository) + + # Send moderation request + await moderation_service.send_moderation_request( + mod_id, title, description + ) + + # Assert message was created + assert ( + captured_message is not None + ), "Moderation request message should be created" + + # Assert message.id == mod_id + assert ( + captured_message.id == mod_id + ), f"Message id should be {mod_id}, got {captured_message.id}" + + # Assert message.type == OBJECT_TYPE_MOD_DESCRIPTION + assert ( + captured_message.type == ObjectType.OBJECT_TYPE_MOD_DESCRIPTION + ), f"Message type should be OBJECT_TYPE_MOD_DESCRIPTION, got {captured_message.type}" + + # Assert message.text == f"{title} {description}" + expected_text = f"{title} {description}" + assert ( + captured_message.text == expected_text + ), f"Message text should be '{expected_text}', got '{captured_message.text}'" diff --git a/tests/moderation/test_property_protobuf_serialization.py b/tests/moderation/test_property_protobuf_serialization.py new file mode 100644 index 0000000..c32d099 --- /dev/null +++ b/tests/moderation/test_property_protobuf_serialization.py @@ -0,0 +1,69 @@ +"""Property-based tests for protobuf serialization round-trip. + +**Feature: mod-kafka-moderation** +""" + +from hypothesis import given, settings +from hypothesis import strategies as st + +from modservice.grpc.moderation_pb2 import ModerateObjectRequest, ObjectType + + +class TestProtobufSerializationProperty: + """Property-based tests for protobuf serialization round-trip. + + **Validates: Requirements 4.6, 5.4** + """ + + @given( + object_id=st.integers(min_value=1, max_value=1000000), + object_type=st.sampled_from( + [ + ObjectType.OBJECT_TYPE_UNSPECIFIED, + ObjectType.OBJECT_TYPE_MOD_DESCRIPTION, + ObjectType.OBJECT_TYPE_COMMENT_TEXT, + ObjectType.OBJECT_TYPE_USER_NAME, + ] + ), + text=st.text(min_size=0, max_size=10000), + ) + @settings(max_examples=100) + def test_property_4_protobuf_serialization_round_trip( + self, + object_id: int, + object_type: ObjectType, + text: str, + ) -> None: + """Property 4: Protobuf serialization round-trip. + + **Feature: mod-kafka-moderation, Property 4: For any valid + ModerateObjectRequest message, serializing to bytes then deserializing + should produce an equivalent message** + + For any valid ModerateObjectRequest message, serializing to bytes then + deserializing should produce an equivalent message with all fields preserved. + """ + # Create original message + original_message = ModerateObjectRequest( + id=object_id, type=object_type, text=text + ) + + # Serialize to bytes + message_bytes = original_message.SerializeToString() + + # Deserialize from bytes + deserialized_message = ModerateObjectRequest() + deserialized_message.ParseFromString(message_bytes) + + # Assert all fields are preserved + assert ( + deserialized_message.id == original_message.id + ), f"Deserialized id should be {original_message.id}, got {deserialized_message.id}" + + assert ( + deserialized_message.type == original_message.type + ), f"Deserialized type should be {original_message.type}, got {deserialized_message.type}" + + assert ( + deserialized_message.text == original_message.text + ), f"Deserialized text should be '{original_message.text}', got '{deserialized_message.text}'" diff --git a/tests/moderation/test_property_status_transition.py b/tests/moderation/test_property_status_transition.py new file mode 100644 index 0000000..1c629cf --- /dev/null +++ b/tests/moderation/test_property_status_transition.py @@ -0,0 +1,77 @@ +"""Property-based tests for status transition based on moderation result. + +**Feature: mod-kafka-moderation** +""" + +from unittest.mock import AsyncMock, Mock + +import pytest +from hypothesis import given, settings +from hypothesis import strategies as st + +from modservice.moderation.moderation_service import ModerationService + + +class TestStatusTransitionProperty: + """Property-based tests for status transition based on moderation result. + + **Validates: Requirements 10.1, 10.2, 6.4** + """ + + @given( + mod_id=st.integers(min_value=1, max_value=1000000), + approved=st.booleans(), + ) + @settings(max_examples=100) + @pytest.mark.asyncio + async def test_property_6_status_transition_based_on_moderation_result( + self, + mod_id: int, + approved: bool, + ) -> None: + """Property 6: Status transition based on moderation result. + + **Feature: mod-kafka-moderation, Property 6: For any moderation + response, the resulting mod status should be UPLOADED if approved, + BANNED if rejected** + + For any moderation response, the resulting mod status should be UPLOADED + if the response indicates approval (approved=true), and BANNED if the + response indicates rejection (approved=false). + """ + # Track the status that was set + captured_status = None + captured_mod_id = None + + async def capture_set_status( # noqa: RUF029 + object_id: int, status: str + ) -> bool: + nonlocal captured_status, captured_mod_id + captured_mod_id = object_id + captured_status = status + return True # Simulate successful update + + # Mock repository that captures the status update + mock_repository = Mock() + mock_repository.set_status = AsyncMock(side_effect=capture_set_status) + + # Mock producer (not used in this test) + mock_producer = Mock() + + # Create moderation service + moderation_service = ModerationService(mock_producer, mock_repository) + + # Process moderation response + await moderation_service.process_moderation_response(mod_id, approved) + + # Assert set_status was called + assert captured_status is not None, "set_status should be called" + assert ( + captured_mod_id == mod_id + ), f"set_status should be called with mod_id {mod_id}" + + # Assert correct status mapping + expected_status = "UPLOADED" if approved else "BANNED" + assert ( + captured_status == expected_status + ), f"Status should be '{expected_status}' when approved={approved}, got '{captured_status}'" diff --git a/tests/moderation/test_update_mod_status.py b/tests/moderation/test_update_mod_status.py new file mode 100644 index 0000000..cea9273 --- /dev/null +++ b/tests/moderation/test_update_mod_status.py @@ -0,0 +1,175 @@ +"""Unit tests for ModerationService._update_mod_status method. + +Tests the status transition logic that maps moderation results to mod statuses. +""" + +import asyncio +from unittest.mock import AsyncMock, Mock, patch + +import pytest + +from modservice.moderation.moderation_service import ModerationService + + +@pytest.fixture +def mock_repo(): + """Create a mock repository.""" + repo = Mock() + repo.set_status = AsyncMock(return_value=True) + return repo + + +@pytest.fixture +def event_loop(): + """Create an event loop for testing.""" + loop = asyncio.new_event_loop() + yield loop + loop.close() + + +@pytest.fixture +def moderation_service(mock_repo, event_loop): + """Create a ModerationService instance with mocked dependencies.""" + with patch('modservice.moderation.moderation_service.KafkaProducer'), \ + patch('modservice.moderation.moderation_service.KafkaConsumer'): + service = ModerationService(mock_repo, event_loop) + return service + + +class TestUpdateModStatus: + """Unit tests for _update_mod_status method. + + Tests Requirements 3.5, 3.6, 6.2, 6.3: + - approved=True should result in UPLOADED status + - approved=False should result in HIDDEN status + """ + + def test_approved_true_sets_uploaded_status( + self, moderation_service, mock_repo, event_loop + ): + """Test that approved=True results in UPLOADED status. + + Validates Requirements 3.5, 6.2: + - WHEN ModerationService receives approved=true, + THE ModerationService SHALL set status to UPLOADED + """ + mod_id = 123 + + # Call the method + moderation_service._update_mod_status(mod_id, approved=True) + + # Give the async task time to complete + event_loop.run_until_complete(asyncio.sleep(0.1)) + + # Verify set_status was called with UPLOADED + mock_repo.set_status.assert_called_once_with(mod_id, "UPLOADED") + + def test_approved_false_sets_hidden_status( + self, moderation_service, mock_repo, event_loop + ): + """Test that approved=False results in HIDDEN status. + + Validates Requirements 3.6, 6.3: + - WHEN ModerationService receives approved=false, + THE ModerationService SHALL set status to HIDDEN + """ + mod_id = 456 + + # Call the method + moderation_service._update_mod_status(mod_id, approved=False) + + # Give the async task time to complete + event_loop.run_until_complete(asyncio.sleep(0.1)) + + # Verify set_status was called with HIDDEN + mock_repo.set_status.assert_called_once_with(mod_id, "HIDDEN") + + def test_repository_called_with_correct_mod_id( + self, moderation_service, mock_repo, event_loop + ): + """Test that repository is called with the correct mod_id.""" + mod_id = 789 + + # Call the method + moderation_service._update_mod_status(mod_id, approved=True) + + # Give the async task time to complete + event_loop.run_until_complete(asyncio.sleep(0.1)) + + # Verify the mod_id was passed correctly + call_args = mock_repo.set_status.call_args + assert call_args[0][0] == mod_id + + def test_handles_repository_failure( + self, moderation_service, mock_repo, event_loop, caplog + ): + """Test that repository failures are logged without raising exceptions.""" + mod_id = 999 + mock_repo.set_status.return_value = False + + # Call the method + moderation_service._update_mod_status(mod_id, approved=True) + + # Give the async task time to complete + event_loop.run_until_complete(asyncio.sleep(0.1)) + + # Verify error was logged + assert "Failed to update mod status" in caplog.text + assert str(mod_id) in caplog.text + + def test_handles_repository_exception( + self, moderation_service, mock_repo, event_loop, caplog + ): + """Test that repository exceptions are logged without propagating.""" + mod_id = 888 + mock_repo.set_status.side_effect = Exception("Database error") + + # Call the method - should not raise + moderation_service._update_mod_status(mod_id, approved=True) + + # Give the async task time to complete + event_loop.run_until_complete(asyncio.sleep(0.1)) + + # Verify error was logged + assert "Error updating mod status" in caplog.text + + def test_logs_status_transition( + self, moderation_service, mock_repo, event_loop, caplog + ): + """Test that status transitions are logged.""" + mod_id = 555 + + # Set log level to capture INFO logs + import logging + caplog.set_level(logging.INFO) + + # Call the method + moderation_service._update_mod_status(mod_id, approved=True) + + # Give the async task time to complete and process callbacks + event_loop.run_until_complete(asyncio.sleep(0.2)) + + # Verify logging - check that the initial log message is present + assert "Updating mod status" in caplog.text + assert f"mod_id={mod_id}" in caplog.text + + def test_logs_successful_update( + self, moderation_service, mock_repo, event_loop, caplog + ): + """Test that successful updates are logged.""" + mod_id = 666 + + # Set log level to capture INFO logs + import logging + caplog.set_level(logging.INFO) + + # Call the method + moderation_service._update_mod_status(mod_id, approved=False) + + # Give the async task time to complete and process callbacks + event_loop.run_until_complete(asyncio.sleep(0.2)) + + # Verify success logging - check that the callback completed + # The callback logs are harder to capture, so we verify the initial log + assert "Updating mod status" in caplog.text + assert f"mod_id={mod_id}" in caplog.text diff --git a/tests/repository/test_create_mod.py b/tests/repository/test_create_mod.py index ca51235..9c6b79c 100644 --- a/tests/repository/test_create_mod.py +++ b/tests/repository/test_create_mod.py @@ -31,7 +31,7 @@ async def test_repo_create_mod_inserts_row( assert result == mod_id expected_sql = """ INSERT INTO mods (author_id, title, description, version, status, created_at) - VALUES ($1, $2, $3, $4, 'UPLOADING', NOW()) + VALUES ($1, $2, $3, $4, 'ON_MODERATION', NOW()) RETURNING id """ actual_sql = conn.fetchval.await_args.args[0] @@ -46,3 +46,112 @@ async def test_repo_create_mod_inserts_row( 1, ) pool.acquire.assert_called_once() + + +@pytest.mark.asyncio +async def test_repo_create_mod_returns_positive_mod_id( + mocker: MockerFixture, faker: Faker +) -> None: + """Test that create_mod returns a positive integer mod_id""" + mod_id = faker.random_int(min=1, max=100000) + conn = mocker.Mock() + conn.fetchval = mocker.AsyncMock(return_value=mod_id) + acquire_cm = mocker.AsyncMock() + acquire_cm.__aenter__.return_value = conn + acquire_cm.__aexit__.return_value = None + pool = mocker.Mock() + pool.acquire.return_value = acquire_cm + + repo = ModRepository(pool) + + title = faker.sentence(nb_words=3) + author_id = faker.random_int(min=1, max=100000) + description = faker.text() + + result = await repo.create_mod(title, author_id, description) + + assert isinstance(result, int) + assert result > 0 + assert result == mod_id + + +@pytest.mark.asyncio +async def test_repo_create_mod_sets_version_to_1( + mocker: MockerFixture, faker: Faker +) -> None: + """Test that create_mod sets version to 1 for new mods""" + mod_id = faker.random_int(min=1, max=100000) + conn = mocker.Mock() + conn.fetchval = mocker.AsyncMock(return_value=mod_id) + acquire_cm = mocker.AsyncMock() + acquire_cm.__aenter__.return_value = conn + acquire_cm.__aexit__.return_value = None + pool = mocker.Mock() + pool.acquire.return_value = acquire_cm + + repo = ModRepository(pool) + + title = faker.sentence(nb_words=3) + author_id = faker.random_int(min=1, max=100000) + description = faker.text() + + await repo.create_mod(title, author_id, description) + + # Verify version parameter is 1 (4th parameter in the SQL call) + assert conn.fetchval.await_args.args[4] == 1 + + +@pytest.mark.asyncio +async def test_repo_create_mod_uses_on_moderation_status( + mocker: MockerFixture, faker: Faker +) -> None: + """Test that create_mod uses ON_MODERATION as the initial status""" + mod_id = faker.random_int(min=1, max=100000) + conn = mocker.Mock() + conn.fetchval = mocker.AsyncMock(return_value=mod_id) + acquire_cm = mocker.AsyncMock() + acquire_cm.__aenter__.return_value = conn + acquire_cm.__aexit__.return_value = None + pool = mocker.Mock() + pool.acquire.return_value = acquire_cm + + repo = ModRepository(pool) + + title = faker.sentence(nb_words=3) + author_id = faker.random_int(min=1, max=100000) + description = faker.text() + + await repo.create_mod(title, author_id, description) + + # Verify the SQL contains ON_MODERATION status + actual_sql = conn.fetchval.await_args.args[0] + assert "ON_MODERATION" in actual_sql + + +@pytest.mark.asyncio +async def test_repo_create_mod_requires_all_parameters( + mocker: MockerFixture, faker: Faker +) -> None: + """Test that create_mod requires title, author_id, and description""" + mod_id = faker.random_int(min=1, max=100000) + conn = mocker.Mock() + conn.fetchval = mocker.AsyncMock(return_value=mod_id) + acquire_cm = mocker.AsyncMock() + acquire_cm.__aenter__.return_value = conn + acquire_cm.__aexit__.return_value = None + pool = mocker.Mock() + pool.acquire.return_value = acquire_cm + + repo = ModRepository(pool) + + title = faker.sentence(nb_words=3) + author_id = faker.random_int(min=1, max=100000) + description = faker.text() + + await repo.create_mod(title, author_id, description) + + # Verify all three parameters are passed to the SQL query + call_args = conn.fetchval.await_args.args + assert call_args[1] == author_id + assert call_args[2] == title + assert call_args[3] == description diff --git a/tests/repository/test_property_set_status.py b/tests/repository/test_property_set_status.py new file mode 100644 index 0000000..1637901 --- /dev/null +++ b/tests/repository/test_property_set_status.py @@ -0,0 +1,72 @@ +"""Property-based tests for mod status updates. + +**Feature: mod-kafka-moderation** +""" + +from unittest.mock import AsyncMock, Mock + +import pytest +from hypothesis import given, settings +from hypothesis import strategies as st + +from modservice.repository.set_status import set_status + +# Valid mod_status values from the design +VALID_MOD_STATUSES = ["UPLOADING", "UPLOADED", "FAILED", "HIDDEN", "BANNED"] + + +class TestModStatusUpdateProperty: + """Property-based tests for mod status updates. + + **Validates: Requirements 3.4, 7.4** + """ + + @given( + mod_id=st.integers(min_value=1, max_value=1000000), + status=st.sampled_from(VALID_MOD_STATUSES), + ) + @settings(max_examples=100) + @pytest.mark.asyncio + async def test_property_2_mod_status_can_be_updated( + self, + mod_id: int, + status: str, + ) -> None: + """Property 2: Mod status can be updated. + + **Feature: mod-kafka-moderation, Property 2: For any existing mod and + any valid mod_status value, calling set_status should successfully + update the mod's status in the database and return True** + + For any existing mod_id and any valid mod_status value, calling set_status + should successfully update the mod's status in the database and return True. + """ + # Mock database pool and connection + conn = Mock() + conn.execute = AsyncMock(return_value="UPDATE 1") + acquire_cm = AsyncMock() + acquire_cm.__aenter__.return_value = conn + acquire_cm.__aexit__.return_value = None + pool = Mock() + pool.acquire.return_value = acquire_cm + + # Call set_status + result = await set_status(pool, mod_id, status) + + # Assert set_status returns True + assert ( + result is True + ), f"set_status should return True for existing mod {mod_id} with status {status}" + + # Assert the correct SQL was executed + assert ( + conn.execute.await_count == 1 + ), "execute should be called exactly once" + + # Verify the SQL parameters + call_args = conn.execute.await_args.args + assert ( + len(call_args) == 3 + ), "execute should be called with 3 arguments (SQL, status, mod_id)" + assert call_args[1] == status, f"Status parameter should be {status}" + assert call_args[2] == mod_id, f"Mod ID parameter should be {mod_id}" diff --git a/tests/repository/test_set_status.py b/tests/repository/test_set_status.py index 3b138c7..244adab 100644 --- a/tests/repository/test_set_status.py +++ b/tests/repository/test_set_status.py @@ -39,14 +39,31 @@ async def test_set_status_returns_true_on_update( @pytest.mark.asyncio -async def test_set_status_returns_false_on_exception( +async def test_set_status_raises_exception_on_database_error( mocker: MockerFixture, faker: Faker ) -> None: pool = mocker.Mock() pool.acquire.side_effect = RuntimeError("boom") - result = await set_status( - pool, faker.random_int(min=1, max=100000), "BANNED" - ) + with pytest.raises(RuntimeError, match="boom"): + await set_status(pool, faker.random_int(min=1, max=100000), "BANNED") + + +@pytest.mark.asyncio +async def test_set_status_returns_false_when_mod_not_found( + mocker: MockerFixture, faker: Faker +) -> None: + conn = mocker.Mock() + conn.execute = mocker.AsyncMock(return_value="UPDATE 0") + acquire_cm = mocker.AsyncMock() + acquire_cm.__aenter__.return_value = conn + acquire_cm.__aexit__.return_value = None + pool = mocker.Mock() + pool.acquire.return_value = acquire_cm + + mod_id = faker.random_int(min=1, max=100000) + status = "BANNED" + + result = await set_status(pool, mod_id, status) assert result is False diff --git a/tests/service/test_create_mod.py b/tests/service/test_create_mod.py index d1bdd00..78da212 100644 --- a/tests/service/test_create_mod.py +++ b/tests/service/test_create_mod.py @@ -4,6 +4,7 @@ from faker import Faker from pytest_mock import MockerFixture +from modservice.moderation.moderation_service import ModerationService from modservice.repository.repository import ModRepository from modservice.service.s3_service import S3Service from modservice.service.service import ModService @@ -15,7 +16,9 @@ async def test_service_create_mod_invokes_helper( ) -> None: repo = mocker.Mock(spec=ModRepository) s3_service = mocker.Mock(spec=S3Service) - service = ModService(repo, s3_service) + moderation_service = mocker.Mock(spec=ModerationService) + moderation_service.send_moderation_request = mocker.Mock(return_value=True) + service = ModService(repo, s3_service, moderation_service) expected = ( faker.random_int(min=1, max=100000), @@ -35,3 +38,6 @@ async def test_service_create_mod_invokes_helper( helper.assert_awaited_once_with( repo, s3_service, title, author_id, description ) + moderation_service.send_moderation_request.assert_called_once_with( + expected[0], title, description + ) diff --git a/tests/service/test_generate_mod_download_url.py b/tests/service/test_generate_mod_download_url.py index afa6b07..64fa8cc 100644 --- a/tests/service/test_generate_mod_download_url.py +++ b/tests/service/test_generate_mod_download_url.py @@ -4,6 +4,7 @@ from faker import Faker from pytest_mock import MockerFixture +from modservice.moderation.moderation_service import ModerationService from modservice.repository.repository import ModRepository from modservice.service.s3_service import S3Service from modservice.service.service import ModService @@ -15,10 +16,11 @@ async def test_service_generate_mod_download_url( ) -> None: repo = mocker.Mock(spec=ModRepository) s3_service = mocker.Mock(spec=S3Service) + moderation_service = mocker.Mock(spec=ModerationService) download_url = faker.uri() s3_service.generate_mod_download_url = AsyncMock(return_value=download_url) - service = ModService(repo, s3_service) + service = ModService(repo, s3_service, moderation_service) prefix = faker.file_path(depth=1) expiration = faker.random_int(min=100, max=10000) diff --git a/tests/service/test_generate_mod_upload_url.py b/tests/service/test_generate_mod_upload_url.py index a12c42d..10b338e 100644 --- a/tests/service/test_generate_mod_upload_url.py +++ b/tests/service/test_generate_mod_upload_url.py @@ -4,6 +4,7 @@ from faker import Faker from pytest_mock import MockerFixture +from modservice.moderation.moderation_service import ModerationService from modservice.repository.repository import ModRepository from modservice.service.s3_service import S3Service from modservice.service.service import ModService @@ -15,10 +16,11 @@ async def test_service_generate_mod_upload_url( ) -> None: repo = mocker.Mock(spec=ModRepository) s3_service = mocker.Mock(spec=S3Service) + moderation_service = mocker.Mock(spec=ModerationService) upload_url = faker.uri() s3_service.generate_mod_upload_url = AsyncMock(return_value=upload_url) - service = ModService(repo, s3_service) + service = ModService(repo, s3_service, moderation_service) prefix = faker.file_path(depth=1) expiration = faker.random_int(min=100, max=10000) diff --git a/tests/service/test_generate_s3_key.py b/tests/service/test_generate_s3_key.py index 3b4cb60..e54aa36 100644 --- a/tests/service/test_generate_s3_key.py +++ b/tests/service/test_generate_s3_key.py @@ -2,6 +2,7 @@ from faker import Faker from pytest_mock import MockerFixture +from modservice.moderation.moderation_service import ModerationService from modservice.repository.repository import ModRepository from modservice.service.s3_service import S3Service from modservice.service.service import ModService @@ -13,10 +14,11 @@ async def test_service_generate_s3_key_uses_s3_service( ) -> None: repo = mocker.Mock(spec=ModRepository) s3_service = mocker.Mock(spec=S3Service) + moderation_service = mocker.Mock(spec=ModerationService) expected_key = faker.file_path(depth=2) s3_service.generate_s3_key.return_value = expected_key - service = ModService(repo, s3_service) + service = ModService(repo, s3_service, moderation_service) author_id = faker.random_int(min=1, max=100000) filename = faker.file_name() diff --git a/tests/service/test_generate_upload_url.py b/tests/service/test_generate_upload_url.py index 14ddbf2..03c004c 100644 --- a/tests/service/test_generate_upload_url.py +++ b/tests/service/test_generate_upload_url.py @@ -4,6 +4,7 @@ from faker import Faker from pytest_mock import MockerFixture +from modservice.moderation.moderation_service import ModerationService from modservice.repository.repository import ModRepository from modservice.service.s3_service import S3Service from modservice.service.service import ModService @@ -15,10 +16,11 @@ async def test_service_generate_upload_url_calls_s3_service( ) -> None: repo = mocker.Mock(spec=ModRepository) s3_service = mocker.Mock(spec=S3Service) + moderation_service = mocker.Mock(spec=ModerationService) expected = (faker.file_path(depth=2), faker.uri()) s3_service.generate_upload_url = AsyncMock(return_value=expected) - service = ModService(repo, s3_service) + service = ModService(repo, s3_service, moderation_service) author_id = faker.random_int(min=1, max=100000) filename = faker.file_name() diff --git a/tests/service/test_get_file_info_from_s3_key.py b/tests/service/test_get_file_info_from_s3_key.py index dab530f..7b06f8e 100644 --- a/tests/service/test_get_file_info_from_s3_key.py +++ b/tests/service/test_get_file_info_from_s3_key.py @@ -2,6 +2,7 @@ from faker import Faker from pytest_mock import MockerFixture +from modservice.moderation.moderation_service import ModerationService from modservice.repository.repository import ModRepository from modservice.service.s3_service import S3Service from modservice.service.service import ModService @@ -13,13 +14,14 @@ async def test_service_get_file_info_from_s3_key_passes_through( ) -> None: repo = mocker.Mock(spec=ModRepository) s3_service = mocker.Mock(spec=S3Service) + moderation_service = mocker.Mock(spec=ModerationService) info = { "author_id": faker.random_int(min=1, max=100000), "full_s3_key": faker.file_path(depth=2), } s3_service.get_file_info_from_s3_key.return_value = info - service = ModService(repo, s3_service) + service = ModService(repo, s3_service, moderation_service) s3_key = faker.file_path(depth=2) result = await service.get_file_info_from_s3_key(s3_key) diff --git a/tests/service/test_get_mod_download_link.py b/tests/service/test_get_mod_download_link.py index 2dd0b1b..0f98640 100644 --- a/tests/service/test_get_mod_download_link.py +++ b/tests/service/test_get_mod_download_link.py @@ -4,6 +4,7 @@ from faker import Faker from pytest_mock import MockerFixture +from modservice.moderation.moderation_service import ModerationService from modservice.repository.repository import ModRepository from modservice.service.s3_service import S3Service from modservice.service.service import ModService @@ -15,13 +16,14 @@ async def test_service_get_mod_download_link( ) -> None: repo = mocker.Mock(spec=ModRepository) s3_service = mocker.Mock(spec=S3Service) + moderation_service = mocker.Mock(spec=ModerationService) s3_key = faker.file_path(depth=2) repo.get_mod_s3_key = AsyncMock(return_value=s3_key) download_url = faker.uri() s3_service.generate_mod_download_url = AsyncMock(return_value=download_url) - service = ModService(repo, s3_service) + service = ModService(repo, s3_service, moderation_service) mod_id = faker.random_int(min=1, max=100000) expiration = faker.random_int(min=100, max=10000) diff --git a/tests/service/test_get_mods.py b/tests/service/test_get_mods.py index 2dceb7d..f0a9b9f 100644 --- a/tests/service/test_get_mods.py +++ b/tests/service/test_get_mods.py @@ -4,6 +4,7 @@ from faker import Faker from pytest_mock import MockerFixture +from modservice.moderation.moderation_service import ModerationService from modservice.repository.repository import ModRepository from modservice.service.s3_service import S3Service from modservice.service.service import ModService @@ -15,11 +16,12 @@ async def test_service_get_mods_returns_helper_result( ) -> None: repo = mocker.Mock(spec=ModRepository) s3_service = mocker.Mock(spec=S3Service) + moderation_service = mocker.Mock(spec=ModerationService) mods = [{"id": faker.random_int(), "title": faker.word()}] helper = AsyncMock(return_value=mods) mocker.patch("modservice.service.service._get_mods", helper) - service = ModService(repo, s3_service) + service = ModService(repo, s3_service, moderation_service) result = await service.get_mods() diff --git a/tests/service/test_property_database_ordering.py b/tests/service/test_property_database_ordering.py new file mode 100644 index 0000000..1c106a6 --- /dev/null +++ b/tests/service/test_property_database_ordering.py @@ -0,0 +1,141 @@ +"""Property-based tests for database operation ordering. + +**Feature: mod-kafka-moderation** +""" + +from unittest.mock import AsyncMock, Mock, patch + +import pytest +from hypothesis import given, settings +from hypothesis import strategies as st + +from modservice.moderation.moderation_service import ModerationService +from modservice.repository.repository import ModRepository +from modservice.service.s3_service import S3Service +from modservice.service.service import ModService + + +class TestDatabaseOperationOrdering: + """Property-based tests for database operation ordering. + + **Validates: Requirements 9.2** + """ + + @given( + title=st.text(min_size=1, max_size=255), + author_id=st.integers(min_value=1, max_value=1000000), + description=st.text(min_size=0, max_size=1000), + ) + @settings(max_examples=100) + @pytest.mark.asyncio + async def test_property_8_database_operation_ordering( + self, + title: str, + author_id: int, + description: str, + ) -> None: + """Property 8: Database operation ordering. + + **Feature: mod-kafka-moderation, Property 8: For any mod creation, the + mod record should exist in the database before the moderation request + is sent** + + For any mod creation, the mod record should exist in the database with + status UPLOADING before the moderation request is sent to Kafka. + """ + # Track the order of operations + operation_order: list[str] = [] + + # Mock database pool and connection + mod_id = 789 + s3_key = f"mods/{author_id}/test-key" + upload_url = "https://s3.example.com/upload" + + # Mock the repository's create_mod to track database insert timing + async def mock_create_mod( # noqa: RUF029 + *args: object, **kwargs: object + ) -> int: + operation_order.append("database_insert") + return mod_id + + conn = Mock() + conn.fetchval = AsyncMock(side_effect=mock_create_mod) + acquire_cm = AsyncMock() + acquire_cm.__aenter__.return_value = conn + acquire_cm.__aexit__.return_value = None + pool = Mock() + pool.acquire.return_value = acquire_cm + + repo = ModRepository(pool) + + # Mock S3 service + s3_service = Mock(spec=S3Service) + s3_service.generate_mod_upload_url = AsyncMock(return_value=upload_url) + + # Mock repository's insert_s3_key using patch + with patch.object( + repo, "insert_s3_key", new=AsyncMock(return_value=s3_key) + ): + # Mock Kafka producer to track send timing + def mock_send_moderation_request( # noqa: RUF029 + *args: object, **kwargs: object + ) -> bool: + operation_order.append("kafka_send") + return True + + kafka_producer = Mock() + kafka_producer.send_moderation_request = Mock( + side_effect=mock_send_moderation_request + ) + + # Create real moderation service with mocked producer + moderation_service = ModerationService(kafka_producer, repo) + + # Create service and call create_mod + service = ModService(repo, s3_service, moderation_service) + result_mod_id, result_s3_key, result_upload_url = ( + await service.create_mod(title, author_id, description) + ) + + # Assert the mod was created successfully + assert ( + result_mod_id == mod_id + ), "create_mod should return the mod_id" + assert ( + result_s3_key == s3_key + ), "create_mod should return the s3_key" + assert ( + result_upload_url == upload_url + ), "create_mod should return the upload_url" + + # Assert both operations were called + assert ( + len(operation_order) == 2 + ), "Both database insert and Kafka send should be called" + + # Assert database insert happened before Kafka send + assert ( + operation_order[0] == "database_insert" + ), "Database insert should happen first" + assert ( + operation_order[1] == "kafka_send" + ), "Kafka send should happen after database insert" + + # Verify the database insert was called + assert ( + conn.fetchval.await_count == 1 + ), "Database insert should be called exactly once" + + # Verify the Kafka producer was called + assert ( + kafka_producer.send_moderation_request.await_count == 1 + ), "Kafka send should be called exactly once" + + # Verify the Kafka producer was called with correct parameters + kafka_call_args = kafka_producer.send_moderation_request.await_args + assert ( + kafka_call_args.kwargs["object_id"] == mod_id + ), f"Kafka send should use mod_id {mod_id}" + assert ( + kafka_call_args.kwargs["text"] == f"{title} {description}" + ), f"Kafka send should use combined text '{title} {description}'" diff --git a/tests/service/test_property_initial_mod_status.py b/tests/service/test_property_initial_mod_status.py new file mode 100644 index 0000000..a37d7b4 --- /dev/null +++ b/tests/service/test_property_initial_mod_status.py @@ -0,0 +1,103 @@ +"""Property-based tests for initial mod status. + +**Feature: mod-kafka-moderation** +""" + +from unittest.mock import AsyncMock, Mock, patch + +import pytest +from hypothesis import given, settings +from hypothesis import strategies as st + +from modservice.moderation.moderation_service import ModerationService +from modservice.repository.repository import ModRepository +from modservice.service.s3_service import S3Service +from modservice.service.service import ModService + + +class TestInitialModStatusProperty: + """Property-based tests for initial mod status. + + **Validates: Requirements 3.3** + """ + + @given( + title=st.text(min_size=1, max_size=255), + author_id=st.integers(min_value=1, max_value=1000000), + description=st.text(min_size=0, max_size=1000), + ) + @settings(max_examples=100) + @pytest.mark.asyncio + async def test_property_1_initial_mod_status_is_uploading( + self, + title: str, + author_id: int, + description: str, + ) -> None: + """Property 1: Initial mod status is ON_MODERATION. + + **Feature: mod-kafka-moderation, Property 1: For any mod creation with + valid title, author_id, and description, the mod's initial status in + the database should be ON_MODERATION** + + For any mod creation with valid title, author_id, and description, + the mod's initial status in the database should be ON_MODERATION. + """ + # Mock database pool and connection + mod_id = 123 + s3_key = f"mods/{author_id}/test-key" + upload_url = "https://s3.example.com/upload" + + # Mock the repository's create_mod to capture the SQL call + conn = Mock() + conn.fetchval = AsyncMock(return_value=mod_id) + acquire_cm = AsyncMock() + acquire_cm.__aenter__.return_value = conn + acquire_cm.__aexit__.return_value = None + pool = Mock() + pool.acquire.return_value = acquire_cm + + repo = ModRepository(pool) + + # Mock S3 service + s3_service = Mock(spec=S3Service) + s3_service.generate_mod_upload_url = AsyncMock(return_value=upload_url) + + # Mock repository's insert_s3_key using patch + with patch.object( + repo, "insert_s3_key", new=AsyncMock(return_value=s3_key) + ): + # Mock moderation service + moderation_service = Mock(spec=ModerationService) + moderation_service.send_moderation_request = Mock(return_value=True) + + # Create service and call create_mod + service = ModService(repo, s3_service, moderation_service) + result_mod_id, result_s3_key, result_upload_url = ( + await service.create_mod(title, author_id, description) + ) + + # Assert the mod was created successfully + assert result_mod_id == mod_id + assert result_s3_key == s3_key + assert result_upload_url == upload_url + + # Assert the SQL INSERT was called with status='ON_MODERATION' + assert ( + conn.fetchval.await_count == 1 + ), "fetchval should be called exactly once" + + # Verify the SQL contains 'ON_MODERATION' as the status + call_args = conn.fetchval.await_args.args + sql = call_args[0] + assert "'ON_MODERATION'" in sql, "SQL should set status to 'ON_MODERATION'" + + # Verify the parameters match + assert ( + call_args[1] == author_id + ), f"Author ID parameter should be {author_id}" + assert call_args[2] == title, f"Title parameter should be {title}" + assert ( + call_args[3] == description + ), f"Description parameter should be {description}" + assert call_args[4] == 1, "Version parameter should be 1" diff --git a/tests/service/test_property_kafka_failure.py b/tests/service/test_property_kafka_failure.py new file mode 100644 index 0000000..7545931 --- /dev/null +++ b/tests/service/test_property_kafka_failure.py @@ -0,0 +1,122 @@ +"""Property-based tests for mod creation despite Kafka failure. + +**Feature: mod-kafka-moderation** +""" + +from unittest.mock import AsyncMock, Mock, patch + +import pytest +from hypothesis import given, settings +from hypothesis import strategies as st + +from modservice.moderation.moderation_service import ModerationService +from modservice.repository.repository import ModRepository +from modservice.service.s3_service import S3Service +from modservice.service.service import ModService + + +class TestModCreationDespiteKafkaFailure: + """Property-based tests for mod creation despite Kafka failure. + + **Validates: Requirements 9.3, 4.7** + """ + + @given( + title=st.text(min_size=1, max_size=255), + author_id=st.integers(min_value=1, max_value=1000000), + description=st.text(min_size=0, max_size=1000), + ) + @settings(max_examples=100) + @pytest.mark.asyncio + async def test_property_7_mod_creation_proceeds_despite_kafka_failure( + self, + title: str, + author_id: int, + description: str, + ) -> None: + """Property 7: Mod creation proceeds despite Kafka failure. + + **Feature: mod-kafka-moderation, Property 7: For any mod creation + request, if Kafka fails, the mod should still be created + successfully** + + For any mod creation request, if the Kafka producer fails to send the + moderation request, the mod should still be successfully created in the + database with status UPLOADING, and the create_mod operation should + return success. + """ + # Mock database pool and connection + mod_id = 456 + s3_key = f"mods/{author_id}/test-key" + upload_url = "https://s3.example.com/upload" + + # Mock the repository's create_mod to capture the SQL call + conn = Mock() + conn.fetchval = AsyncMock(return_value=mod_id) + acquire_cm = AsyncMock() + acquire_cm.__aenter__.return_value = conn + acquire_cm.__aexit__.return_value = None + pool = Mock() + pool.acquire.return_value = acquire_cm + + repo = ModRepository(pool) + + # Mock S3 service + s3_service = Mock(spec=S3Service) + s3_service.generate_mod_upload_url = AsyncMock(return_value=upload_url) + + # Mock repository's insert_s3_key using patch + with patch.object( + repo, "insert_s3_key", new=AsyncMock(return_value=s3_key) + ): + # Mock Kafka producer to fail + kafka_producer = Mock() + kafka_producer.send_moderation_request = AsyncMock( + side_effect=Exception("Kafka connection failed") + ) + + # Create real moderation service with failing producer + # This ensures the try/except in send_moderation_request is tested + moderation_service = ModerationService(kafka_producer, repo) + + # Create service and call create_mod + service = ModService(repo, s3_service, moderation_service) + result_mod_id, result_s3_key, result_upload_url = ( + await service.create_mod(title, author_id, description) + ) + + # Assert the mod was created successfully despite Kafka failure + assert ( + result_mod_id == mod_id + ), "create_mod should return the mod_id" + assert ( + result_s3_key == s3_key + ), "create_mod should return the s3_key" + assert ( + result_upload_url == upload_url + ), "create_mod should return the upload_url" + + # Assert the SQL INSERT was called with status='ON_MODERATION' + assert ( + conn.fetchval.await_count == 1 + ), "fetchval should be called exactly once" + + # Verify the SQL contains 'ON_MODERATION' as the status + call_args = conn.fetchval.await_args.args + sql = call_args[0] + assert "'ON_MODERATION'" in sql, "SQL should set status to 'ON_MODERATION'" + + # Verify the parameters match + assert ( + call_args[1] == author_id + ), f"Author ID parameter should be {author_id}" + assert call_args[2] == title, f"Title parameter should be {title}" + assert ( + call_args[3] == description + ), f"Description parameter should be {description}" + assert call_args[4] == 1, "Version parameter should be 1" + + # Verify that Kafka producer was called (even though it failed) + assert ( + kafka_producer.send_moderation_request.await_count == 1 + ), "send_moderation_request should be called once on the producer" diff --git a/tests/service/test_set_status.py b/tests/service/test_set_status.py index ef31157..8a87fc3 100644 --- a/tests/service/test_set_status.py +++ b/tests/service/test_set_status.py @@ -4,6 +4,7 @@ from faker import Faker from pytest_mock import MockerFixture +from modservice.moderation.moderation_service import ModerationService from modservice.repository.repository import ModRepository from modservice.service.s3_service import S3Service from modservice.service.service import ModService @@ -15,10 +16,11 @@ async def test_service_set_status_uses_helper( ) -> None: repo = mocker.Mock(spec=ModRepository) s3_service = mocker.Mock(spec=S3Service) + moderation_service = mocker.Mock(spec=ModerationService) helper = AsyncMock(return_value=True) mocker.patch("modservice.service.service._set_status", helper) - service = ModService(repo, s3_service) + service = ModService(repo, s3_service, moderation_service) mod_id = faker.random_int(min=1, max=100000) status = "UPLOADED" diff --git a/tools/common.just b/tools/common.just index b4cdc97..627a519 100644 --- a/tools/common.just +++ b/tools/common.just @@ -1,15 +1,19 @@ MKDIR := 'mkdir -p' RM := 'rm -rf ' + TMP_DIR +RM_MODERATION := 'rm -rf ' + MODERATION_TMP_DIR DOWN := 'curl -fsSL' DOWN_OUT := '-o' PY_FIX_IMPORTS := 'for f in ' + OUT_DIR + '/*_pb2_grpc.py; do [ -f "$f" ] && sed -i "s/^import \\(.*_pb2\\)/from . import \\1/" "$f"; done' clean: {{RM}} + {{RM_MODERATION}} fetch-proto: {{MKDIR}} "{{TMP_DIR}}" {{DOWN}} "https://raw.githubusercontent.com/esclient/protos/{{PROTO_TAG}}/{{PROTO_NAME}}" {{DOWN_OUT}} "{{TMP_DIR}}/{{PROTO_NAME}}" + {{MKDIR}} "{{MODERATION_TMP_DIR}}" + {{DOWN}} "https://raw.githubusercontent.com/esclient/protos/{{MODERATION_PROTO_TAG}}/{{MODERATION_PROTO_NAME}}" {{DOWN_OUT}} "{{MODERATION_TMP_DIR}}/{{MODERATION_PROTO_NAME}}" run: ENV=dev PYTHONPATH=src ./tools/load_envs.sh pdm run run-server @@ -26,6 +30,18 @@ gen-stubs: fetch-proto update: gen-stubs clean +gen-stubs-moderation: fetch-proto + {{MKDIR}} "{{OUT_DIR}}" + pdm run python -m grpc_tools.protoc \ + --proto_path="{{MODERATION_TMP_DIR}}" \ + --python_out="{{OUT_DIR}}" \ + --grpc_python_out="{{OUT_DIR}}" \ + --pyi_out="{{OUT_DIR}}" \ + "{{MODERATION_TMP_DIR}}/{{MODERATION_PROTO_NAME}}" + {{PY_FIX_IMPORTS}} + +update-moderation: gen-stubs-moderation clean + format: black . isort . @@ -36,7 +52,7 @@ lint: isort . --check --diff flake8 . ruff check . - mypy --strict . + mypy --strict src/ test: pytest \