diff --git a/docs/development.md b/docs/development.md index d76c991e4..403550b78 100644 --- a/docs/development.md +++ b/docs/development.md @@ -461,6 +461,103 @@ task docker:purge && EDA_SECRET_KEY=insecure task docker:up Do **not** confuse the two — see the command docstring in `src/aap_eda/core/management/commands/rotate_db_encryption_key.py`. +## Log purge + +- **Submit a job:** A superuser sends `POST /api/eda/v1/logs/purge/`. HTTP 202 + with a `job_id` means dispatcherd accepted the job; deletion may still be + running. +- **Choose a cutoff:** For an age-based purge, supply exactly one of + `before_date` (midnight UTC in `YYYY-MM-DD` format) or a positive ISO 8601 + `retention_duration`. Only rows with `log_timestamp` strictly before the + cutoff qualify, including earlier audit markers. Future UTC cutoffs are + rejected. +- **Purge logs globally:** Send `"purge-all-logs": true` without activation + selectors. Alone, it deletes logs at every timestamp. Combine it with + `before_date` or `retention_duration` to delete only older logs, including + unattached logs. Each affected process receives a new audit marker; unattached + logs cannot receive one. The deleted count includes earlier markers. If + processes keep writing logs, new rows can appear during or after the purge. + Without a cutoff or this flag, the API returns HTTP 400 before queueing. +- **Choose the scope:** With no selectors, qualifying logs from every + process are included, including processes without an activation. + `activation_ids` and `activation_names` select the union of matching + activations. Empty or unknown selectors are rejected before queueing. +- **Choose audit behavior:** Audit markers are enabled by default. Set API + `audit_trail` to `false` to disable them. + +For example, this request queues a purge for the named activation: + +```http +POST /api/eda/v1/logs/purge/ +Content-Type: application/json + +{"retention_duration": "P30D", "audit_trail": true, + "activation_names": ["production-rules"]} +``` + +To replace all existing logs with new audit markers: + +```http +POST /api/eda/v1/logs/purge/ +Content-Type: application/json + +{"purge-all-logs": true} +``` + +To purge logs globally before a cutoff: + +```http +POST /api/eda/v1/logs/purge/ +Content-Type: application/json + +{"purge-all-logs": true, "before_date": "2024-01-01"} +``` + +**CLI options** + +- Use either `--date YYYY-MM-DD` or `--retention-duration P30D`. Calendar + months and years in a duration are subtracted in UTC. If the target month + lacks the original day, the cutoff uses that month's final day. +- Use `--purge-all-logs` alone to delete every existing log, or combine it + with `--date` or `--retention-duration` to delete only older logs globally. + This flag cannot be combined with activation selectors or + `--audit-trail=false`. Omitting both a cutoff and this flag raises a command + error before deletion. +- Use `--activation-ids 123 456`, `--activation-names name1 name2`, or both to + limit deletion to the selected activations. Omit both to include every + activation and processes without an activation. +- Use `--audit-trail=true` or `--audit-trail=false`. A bare `--audit-trail` is + invalid. Default is True. +- The optional `--internal-job-uuid` accepts a UUID or a custom ID of 1-128 + ASCII letters, digits, periods, underscores, or hyphens, starting with a + letter or digit. Valid UUIDs are normalized. The value appears in audit + markers under `Purge Job ID`. The API supplies its task UUID automatically. + +**Deletion and audit markers** + +- Deletion runs once per process. If a later process fails, worker logs record + the count already deleted, and the job reports failure. Audit markers record + the count for each affected process when audit is enabled. +- When audit is enabled, each process that loses log rows gets one marker with + its count for this invocation. Markers use `Recorded at` for the time the + audit row was written and store that time in `log_created_at`. The API + returns this recorded time. +- Cutoff purges treat earlier markers like other log rows: they are deleted and + counted once their `log_timestamp` falls before the cutoff. Recent markers + can therefore remain until a later cutoff purge reaches them. + +**Check the outcome** + +- Inspect default-worker logs by the returned job ID for started, completed, + skipped, and failed outcomes, including `deleted_count`: + +```shell +JOB_ID= +kubectl logs -n "${NAMESPACE:-aap-eda}" \ + -l app=eda,comp=default-worker --all-containers=true --since=24h \ + | grep "job_id=${JOB_ID}" +``` + ### Running linters The project uses `flake8` with a set of plugins, `black`, `isort` and `ruff` (experimental), diff --git a/poetry.lock b/poetry.lock index 1a115d379..d72c53aae 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1,4 +1,4 @@ -# This file is automatically @generated by Poetry 2.4.1 and should not be changed by hand. +# This file is automatically @generated by Poetry 2.3.4 and should not be changed by hand. [[package]] name = "adal" @@ -3648,4 +3648,4 @@ dev = ["psycopg-binary"] [metadata] lock-version = "2.1" python-versions = ">=3.12,<3.13" -content-hash = "9681770ca45339d4af6cdc82588f70fec852a8b305c0d034056d8a767bc12560" +content-hash = "471e617d209875a3d38ef5868a94eec8c6e44696d1d9876d1a247119d36d3214" diff --git a/pyproject.toml b/pyproject.toml index d73aed13a..e4359c62e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -68,6 +68,7 @@ pyjwt = { version = ">=2.13.0,<3", extras = ["crypto"] } ecdsa = "0.18.*" validators = "^0.34.0" django-flags = "^5.0.13" +isodate = "^0.7.2" insights-analytics-collector = "^0.3.2" distro = "^1.9.0" dispatcherd = { version = "v2026.05.27", extras = ["pg_notify"] } diff --git a/src/aap_eda/api/serializers/__init__.py b/src/aap_eda/api/serializers/__init__.py index 9c072c486..ae760392b 100644 --- a/src/aap_eda/api/serializers/__init__.py +++ b/src/aap_eda/api/serializers/__init__.py @@ -21,6 +21,8 @@ ActivationReadSerializer, ActivationSerializer, ActivationUpdateSerializer, + LogPurgeRequestSerializer, + LogPurgeResponseSerializer, PostActivationSerializer, ) from .auth import JWTTokenSerializer, LoginSerializer, RefreshTokenSerializer @@ -114,6 +116,8 @@ "ActivationListSerializer", "ActivationCreateSerializer", "ActivationUpdateSerializer", + "LogPurgeRequestSerializer", + "LogPurgeResponseSerializer", "ActivationReadSerializer", "ActivationCopySerializer", "ActivationInstanceSerializer", diff --git a/src/aap_eda/api/serializers/activation.py b/src/aap_eda/api/serializers/activation.py index 069018994..c710e19aa 100644 --- a/src/aap_eda/api/serializers/activation.py +++ b/src/aap_eda/api/serializers/activation.py @@ -16,6 +16,7 @@ import secrets import uuid from dataclasses import dataclass, field +from datetime import datetime, time, timezone as datetime_timezone from typing import Optional import yaml @@ -54,7 +55,9 @@ get_resolved_secrets, get_secret_fields, ) +from aap_eda.core.utils.delete_log_util import resolve_activation_ids from aap_eda.core.utils.k8s_service_name import create_k8s_service_name +from aap_eda.core.utils.log_retention import retention_cutoff from aap_eda.core.utils.rulebook import ( build_rulebook_with_event_streams, build_source_list, @@ -89,6 +92,139 @@ def _update_event_stream_source(validated_data: dict) -> str: raise InvalidEventStreamSource(e) from e +class LogPurgeRequestSerializer(serializers.Serializer): + """Serializer for the asynchronous log purge endpoint.""" + + before_date = serializers.DateField( + required=False, + help_text="Exclusive UTC midnight cutoff (YYYY-MM-DD).", + ) + retention_duration = serializers.CharField( + required=False, + help_text="Exclusive cutoff as a positive ISO 8601 duration.", + ) + audit_trail = serializers.BooleanField( + required=False, + default=True, + help_text="Write one audit marker per process that loses logs.", + ) + activation_ids = serializers.ListField( + child=serializers.IntegerField(min_value=1), + required=False, + allow_empty=False, + help_text="Limit deletion to these activation IDs.", + ) + activation_names = serializers.ListField( + child=serializers.CharField(allow_blank=False, trim_whitespace=False), + required=False, + allow_empty=False, + help_text="Limit deletion to these activation names.", + ) + + def get_fields(self): + fields = super().get_fields() + fields["purge-all-logs"] = serializers.BooleanField( + source="purge_all_logs", + required=False, + default=False, + help_text=( + "Delete all existing process logs globally. Can be combined " + "with before_date or retention_duration to limit deletion " + "to older logs. Cannot be combined with activation selectors " + "or disabled audit." + ), + ) + return fields + + @staticmethod + def _validate_full_purge(attrs): + """Reject selector or audit-off combinations with purge-all-logs.""" + if "activation_ids" in attrs or "activation_names" in attrs: + raise serializers.ValidationError( + "purge-all-logs cannot be combined with activation selectors." + ) + if not attrs["audit_trail"]: + raise serializers.ValidationError( + "purge-all-logs requires audit_trail=true." + ) + + def _validate_cutoff(self, attrs): + """Ensure exactly one cutoff is supplied and normalize it.""" + before_date = attrs.get("before_date") + duration = attrs.get("retention_duration") + has_date = before_date is not None + has_duration = duration is not None + err = "Exactly one of before_date or retention_duration is required." + + if has_date and has_duration: + raise serializers.ValidationError(err) + if not has_date and not has_duration and not attrs["purge_all_logs"]: + raise serializers.ValidationError(err) + if has_date: + self._validate_before_date(attrs, before_date) + elif has_duration: + self._validate_duration(duration) + + @staticmethod + def _validate_before_date(attrs, before_date): + """Reject future dates and normalise before_date to ISO string.""" + cutoff = datetime.combine( + before_date, time.min, tzinfo=datetime_timezone.utc + ) + if cutoff > datetime.now(datetime_timezone.utc): + raise serializers.ValidationError( + {"before_date": "The UTC cutoff must not be future."} + ) + attrs["before_date"] = before_date.isoformat() + + @staticmethod + def _validate_duration(duration): + """Validate retention_duration as a positive ISO 8601 duration.""" + try: + retention_cutoff(duration) + except ValueError as exc: + raise serializers.ValidationError( + {"retention_duration": str(exc)} + ) from exc + + @staticmethod + def _resolve_activations(attrs): + """Resolve activation names to IDs.""" + if "activation_ids" not in attrs and "activation_names" not in attrs: + return + try: + attrs["activation_ids"] = sorted( + resolve_activation_ids( + attrs.get("activation_ids"), + attrs.get("activation_names"), + ) + ) + except ValueError as exc: + raise serializers.ValidationError(str(exc)) from exc + attrs.pop("activation_names", None) + + def validate(self, attrs): + unknown_fields = set(self.initial_data) - set(self.fields) + if unknown_fields: + raise serializers.ValidationError( + dict.fromkeys(sorted(unknown_fields), "Unknown field.") + ) + + if attrs["purge_all_logs"]: + self._validate_full_purge(attrs) + self._validate_cutoff(attrs) + if attrs["purge_all_logs"]: + return attrs + self._resolve_activations(attrs) + return attrs + + +class LogPurgeResponseSerializer(serializers.Serializer): + """Acknowledgement returned after a purge task is queued.""" + + job_id = serializers.UUIDField() + + def _update_k8s_service_name(validated_data: dict) -> str: service_name = validated_data.get("k8s_service_name") return service_name or create_k8s_service_name(validated_data["name"]) diff --git a/src/aap_eda/api/urls.py b/src/aap_eda/api/urls.py index 83c673d29..555104de6 100644 --- a/src/aap_eda/api/urls.py +++ b/src/aap_eda/api/urls.py @@ -37,6 +37,7 @@ router.register("projects", views.ProjectViewSet) router.register("rulebooks", views.RulebookViewSet) router.register("activations", views.ActivationViewSet) +router.register("logs", views.LogPurgeViewSet, basename="logs") router.register( "activation-instances", views.ActivationInstanceViewSet, diff --git a/src/aap_eda/api/views/__init__.py b/src/aap_eda/api/views/__init__.py index 3b61fdf77..ab76d81d2 100644 --- a/src/aap_eda/api/views/__init__.py +++ b/src/aap_eda/api/views/__init__.py @@ -12,7 +12,11 @@ # See the License for the specific language governing permissions and # limitations under the License. -from .activation import ActivationInstanceViewSet, ActivationViewSet +from .activation import ( + ActivationInstanceViewSet, + ActivationViewSet, + LogPurgeViewSet, +) from .auth import SessionLoginView, SessionLogoutView, TokenRefreshView from .config import ConfigView from .credential_input_source import CredentialInputSourceViewSet @@ -40,6 +44,7 @@ # activations "ActivationViewSet", "ActivationInstanceViewSet", + "LogPurgeViewSet", # user "CurrentUserView", "CurrentUserAwxTokenViewSet", diff --git a/src/aap_eda/api/views/activation.py b/src/aap_eda/api/views/activation.py index 357f7580e..5a2e89a27 100644 --- a/src/aap_eda/api/views/activation.py +++ b/src/aap_eda/api/views/activation.py @@ -31,11 +31,13 @@ from rest_framework.response import Response from aap_eda.api import exceptions as api_exc, filters, serializers +from aap_eda.api.permissions import IsSystemAdmin from aap_eda.api.serializers.activation import is_activation_valid from aap_eda.core import models from aap_eda.core.enums import Action, ActivationStatus, ProcessParentType from aap_eda.core.health import check_dispatcherd_workers_health from aap_eda.core.utils import logging_utils +from aap_eda.tasks.management import submit_log_purge from aap_eda.tasks.orchestrator import ( delete_rulebook_process, restart_rulebook_process, @@ -868,3 +870,54 @@ def logs(self, request, id): results, many=True ) return self.get_paginated_response(serializer.data) + + +class LogPurgeViewSet(viewsets.GenericViewSet): + """Superuser-only log purge scheduling endpoint.""" + + permission_classes = (IsSystemAdmin,) + serializer_class = serializers.LogPurgeRequestSerializer + + @extend_schema( + description=( + "Queue deletion of process logs using one UTC cutoff: " + "before_date (YYYY-MM-DD) or retention_duration (positive ISO " + "8601). Alternatively, purge-all-logs=true deletes logs " + "globally. It can be combined with before_date or " + "retention_duration to delete only older logs, including prior " + "audit markers and logs without a process. It leaves one new " + "audit marker per affected process. Logs without a process " + "cannot receive a marker. This mode cannot be combined with " + "activation selectors or disabled audit. Without activation " + "selectors, all processes, including orphaned ones, are included. " + "Activation IDs and names limit " + "cutoff deletion to their union. The returned job ID " + "acknowledges scheduling only." + ), + request=serializers.LogPurgeRequestSerializer, + responses={ + status.HTTP_202_ACCEPTED: serializers.LogPurgeResponseSerializer, + status.HTTP_400_BAD_REQUEST: OpenApiResponse( + description="The request contains invalid purge options." + ), + status.HTTP_403_FORBIDDEN: OpenApiResponse( + description="Only superusers can queue a log purge." + ), + status.HTTP_500_INTERNAL_SERVER_ERROR: OpenApiResponse( + description="The purge task could not be queued." + ), + }, + ) + @action(methods=["post"], detail=False, url_path="purge") + def purge(self, request): + serializer = self.get_serializer(data=request.data) + serializer.is_valid(raise_exception=True) + try: + job_id = submit_log_purge(**serializer.validated_data) + except Exception: + logger.exception("Unable to queue log purge") + return Response( + {"detail": "Unable to queue log purge job."}, + status=status.HTTP_500_INTERNAL_SERVER_ERROR, + ) + return Response({"job_id": job_id}, status=status.HTTP_202_ACCEPTED) diff --git a/src/aap_eda/core/management/commands/purge_log_records.py b/src/aap_eda/core/management/commands/purge_log_records.py index deb5d8173..5905405d1 100644 --- a/src/aap_eda/core/management/commands/purge_log_records.py +++ b/src/aap_eda/core/management/commands/purge_log_records.py @@ -12,40 +12,49 @@ # See the License for the specific language governing permissions and # limitations under the License. -from datetime import datetime +import re +from argparse import ArgumentTypeError +from datetime import datetime, time, timezone +from uuid import UUID from django.core.management.base import ( BaseCommand, CommandError, CommandParser, ) -from django.db import transaction -from django.db.models import Q -from aap_eda.core import models +from aap_eda.core.utils.delete_log_util import ( + LogPurgeError, + delete_logs_older_than, + purge_unattached_logs, +) +from aap_eda.core.utils.log_retention import retention_cutoff class Command(BaseCommand): - """Purge the logs from a rulebook process.""" + """Purge old logs globally or for selected activations.""" + + VERSION = "1.1.1" help = ( - "Purge log records from rulebook processes. " - "Always purges all log records older than --date globally. " - "Activation ids or names only control detailed reporting for those " - "activations and orphaned records; they do not limit deletion." + "Purge log records older than exactly one cutoff, or use " + "--purge-all-logs with an optional cutoff to delete logs globally " + "and leave one audit marker per affected process. Without activation " + "selectors, include every process (including orphaned processes). " + "Activation IDs and names limit deletion to their union. " + "One audit marker per affected process is enabled by default." ) + def get_version(self): + return self.VERSION + def add_arguments(self, parser: CommandParser) -> None: parser.add_argument( "--activation-ids", nargs="+", type=int, dest="activation-ids", - help=( - "Activation ids to include in detailed reporting only. " - "This does not limit deletion: all log records older than " - "--date are purged globally (e.g., 123 456)." - ), + help=("Limit deletion to these activation IDs (e.g., 123 456)."), ) parser.add_argument( "--activation-names", @@ -53,110 +62,229 @@ def add_arguments(self, parser: CommandParser) -> None: type=str, dest="activation-names", help=( - "Activation names to include in detailed reporting only. " - "This does not limit deletion: all log records older than " - "--date are purged globally (e.g., ActivationName1 " - "ActivationName2)." + "Limit deletion to these activation names (e.g., " + "ActivationName1 ActivationName2)." ), ) parser.add_argument( "--date", dest="date", action="store", - required=True, help=( "Purge records older than this date from the database. " - "The cutoff date in YYYY-MM-DD format" + "The UTC cutoff date in YYYY-MM-DD format. Mutually exclusive " + "with --retention-duration." ), ) - def purge_log_records( - self, ids: list[int], names: list[str], cutoff_timestamp: datetime - ) -> None: - # Global purge of all old logs (regardless of activation) - cutoff_ts = int(cutoff_timestamp.timestamp()) - old_logs = models.RulebookProcessLog.objects.filter( - log_timestamp__lt=cutoff_ts + parser.add_argument( + "--retention-duration", + dest="retention_duration", + help=( + "Purge records older than this ISO 8601 duration, for " + "example P30D or P1M. Mutually exclusive with --date." + ), + ) + parser.add_argument( + "--purge-all-logs", + dest="purge-all-logs", + action="store_true", + help=( + "Delete all existing process logs. When used alone, deletes " + "every log at every timestamp. When combined with --date or " + "--retention-duration, deletes logs older than the cutoff " + "across all activations. Also deletes unattached logs " + "(filtered by cutoff when provided). Cannot be combined " + "with activation selectors or --audit-trail=false." + ), + ) + parser.add_argument( + "--audit-trail", + dest="audit_trail", + type=self._audit_trail_value, + default=True, + metavar="true|false", + help="Create a marker for each affected process (default: true).", + ) + parser.add_argument( + "--internal-job-uuid", + dest="internal-job-uuid", + default=None, + help=( + "Purge job UUID or custom ID (1-128 letters, digits, periods, " + "underscores, or hyphens) to include in audit markers." + ), ) - deleted_count = old_logs.count() - if deleted_count == 0: - self.stdout.write( - self.style.SUCCESS( - f"No log records found older than " - f"{cutoff_timestamp.strftime('%Y-%m-%d')}." - ) + @staticmethod + def _audit_trail_value(value: str) -> bool: + """Parse audit-trail CLI flag to boolean.""" + if value == "true": + return True + if value == "false": + return False + raise ArgumentTypeError("--audit-trail must be true or false.") + + @staticmethod + def _cutoff(options: dict) -> datetime: + """Parse and validate cutoff date from CLI options.""" + cutoff_date = options.get("date") + retention_duration = options.get("retention_duration") + if cutoff_date is not None and retention_duration is not None: + raise CommandError( + "--date and --retention-duration are mutually exclusive." ) - return - # Delete all old logs globally - old_logs.delete() + if cutoff_date is not None: + try: + parsed_date = datetime.strptime(cutoff_date, "%Y-%m-%d").date() + except (TypeError, ValueError) as exc: + raise CommandError( + "--date must use YYYY-MM-DD format." + ) from exc + if parsed_date.strftime("%Y-%m-%d") != cutoff_date: + raise CommandError("--date must use YYYY-MM-DD format.") + return datetime.combine(parsed_date, time.min, tzinfo=timezone.utc) - self.stdout.write( - self.style.SUCCESS( - f"Purged {deleted_count} log records older than " - f"{cutoff_timestamp.strftime('%Y-%m-%d')} globally." - ) - ) + if retention_duration is not None: + try: + return retention_cutoff(retention_duration) + except ValueError as exc: + raise CommandError(str(exc)) from exc - # Create audit trail logs for reporting - audit_logs = [] + raise CommandError( + "Exactly one of --date or --retention-duration is required." + ) - if not bool(ids) and not bool(names): - # No filters: Create audit logs for all instances that might - # have been affected - instances = models.RulebookProcess.objects.all() + def _purge_mode(self, options: dict) -> tuple[bool, datetime | None]: + """Resolve and validate the purge scope and cutoff from CLI options.""" + ids = options.get("activation-ids") + names = options.get("activation-names") + purge_all_logs = options.get("purge-all-logs", False) + if not isinstance(purge_all_logs, bool): + raise CommandError("--purge-all-logs must be a boolean flag.") + if purge_all_logs and (ids is not None or names is not None): + raise CommandError( + "--purge-all-logs cannot be combined with " + "activation selectors." + ) + has_cutoff = ( + options.get("date") is not None + or options.get("retention_duration") is not None + ) + if purge_all_logs and not has_cutoff: + cutoff = None + elif has_cutoff: + cutoff = self._cutoff(options) else: - # Filters provided: Create audit logs only for specified - # activations and orphaned records - instances = models.RulebookProcess.objects.filter( - Q(activation__id__in=ids) - | Q(activation__name__in=names) - | Q(activation__isnull=True), + raise CommandError( + "Exactly one of --date, --retention-duration, or " + "--purge-all-logs is required." ) + if cutoff is not None and cutoff > datetime.now(timezone.utc): + raise CommandError("--date must not be in the future.") + return purge_all_logs, cutoff - # Report on orphaned records for user visibility - orphaned_count = models.RulebookProcess.objects.filter( - activation__isnull=True - ).count() - if orphaned_count > 0: - self.stdout.write( - f"Audit trail will include {orphaned_count} orphaned " - f"RulebookProcess records (with NULL activation)." - ) + def _validated_audit_trail(self, value, purge_all_logs: bool) -> bool: + """Coerce and validate the audit-trail flag.""" + if isinstance(value, str): + try: + value = self._audit_trail_value(value) + except ArgumentTypeError as exc: + raise CommandError(str(exc)) from exc + elif not isinstance(value, bool): + raise CommandError("--audit-trail must be true or false.") + if purge_all_logs and not value: + raise CommandError("--purge-all-logs requires audit trail.") + return value - # Create audit trail logs for instances - dt = datetime.now().strftime("%Y-%m-%d %H:%M:%S") - for instance in instances: - audit_logs.append( - models.RulebookProcessLog( - log=( - f"All log records older than " - f"{cutoff_timestamp.strftime('%Y-%m-%d')} " - f"were purged at {dt}." - ), - activation_instance_id=instance.id, - log_timestamp=cutoff_ts, - ) - ) + @staticmethod + def _delete_unattached_logs( + deleted_count: int, cutoff_timestamp: float | None + ) -> int: + """Delete logs not linked to any activation instance.""" + try: + deleted_count += purge_unattached_logs(cutoff_timestamp) + except Exception as exc: + raise LogPurgeError( + f"Log purge failed after {deleted_count} committed rows: " + f"{exc}", + deleted_count=deleted_count, + ) from exc + return deleted_count - if audit_logs: - models.RulebookProcessLog.objects.bulk_create(audit_logs) - self.stdout.write( - f"Created {len(audit_logs)} audit trail log entries." - ) + @staticmethod + def _validated_job_uuid(job_uuid): + """Normalise a UUID or validate a custom job ID string.""" + if job_uuid is None: + return None + try: + return str(UUID(str(job_uuid))) + except (TypeError, ValueError, AttributeError) as exc: + if not isinstance(job_uuid, str) or not re.fullmatch( + r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}", job_uuid + ): + raise CommandError( + "--internal-job-uuid must be a UUID or a custom ID " + "of 1-128 ASCII letters, digits, periods, " + "underscores, or hyphens, starting with a letter " + "or digit." + ) from exc + return job_uuid - @transaction.atomic def handle(self, *args, **options): - input_ids = options.get("activation-ids") or [] - input_names = options.get("activation-names") or [] - cutoff_date = options.get("date") + ids = options.get("activation-ids") + names = options.get("activation-names") + purge_all_logs, cutoff = self._purge_mode(options) + job_uuid = self._validated_job_uuid(options.get("internal-job-uuid")) + audit_trail = self._validated_audit_trail( + options["audit_trail"], purge_all_logs + ) try: - ts = datetime.strptime(cutoff_date, "%Y-%m-%d") - except ValueError as e: - raise CommandError(f"{e}") from e + process_deleted_count = delete_logs_older_than( + cutoff, + activation_ids=ids, + activation_names=names, + audit_trail=audit_trail, + internal_job_uuid=job_uuid, + purge_all_logs=purge_all_logs, + ) + if purge_all_logs: + deleted_count = self._delete_unattached_logs( + process_deleted_count, + cutoff.timestamp() if cutoff is not None else None, + ) + except (LogPurgeError, ValueError) as exc: + raise CommandError(str(exc)) from exc - self.purge_log_records( - ids=input_ids, names=input_names, cutoff_timestamp=ts + if purge_all_logs: + marker_result = ( + "Created audit markers for affected processes." + if process_deleted_count + else "No audit markers created." + ) + noun = "record" if deleted_count == 1 else "records" + tag = f"[purge_log_records v{self.VERSION}]" + if cutoff is not None: + cutoff_str = cutoff.strftime("%Y-%m-%d %H:%M:%S UTC") + return self.style.SUCCESS( + f"Purged {deleted_count} log {noun} globally older " + f"than {cutoff_str}. {marker_result} {tag}" + ) + return self.style.SUCCESS( + f"Purged {deleted_count} log {noun} globally at all " + f"timestamps. {marker_result} {tag}" + ) + + tag = f"[purge_log_records v{self.VERSION}]" + noun = "record" if process_deleted_count == 1 else "records" + scope = ( + "globally" + if ids is None and names is None + else "for selected activations" + ) + return self.style.SUCCESS( + f"Purged {process_deleted_count} log {noun} older than " + f"{cutoff.strftime('%Y-%m-%d %H:%M:%S UTC')} {scope}. {tag}" ) diff --git a/src/aap_eda/core/utils/delete_log_util.py b/src/aap_eda/core/utils/delete_log_util.py new file mode 100644 index 000000000..1828b8db7 --- /dev/null +++ b/src/aap_eda/core/utils/delete_log_util.py @@ -0,0 +1,284 @@ +# Copyright 2026 Red Hat, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from dataclasses import dataclass +from datetime import datetime, timezone +from math import ceil + +from django.db import transaction +from django.db.models import Max + +from aap_eda.core import models + +PROCESS_PAGE_SIZE = 1000 + + +def _utc_text(value: datetime) -> str: + """Format datetime as UTC string with optional microseconds.""" + value = value.astimezone(timezone.utc) + time_format = "%Y-%m-%d %H:%M:%S" + if value.microsecond: + time_format += ".%f" + return value.strftime(time_format + " UTC") + + +class LogPurgeError(Exception): + """A purge failure that includes the number of committed deletions.""" + + def __init__(self, message: str, deleted_count: int): + self.deleted_count = deleted_count + super().__init__(message) + + +@dataclass(frozen=True) +class _PurgeOptions: + """Validated purge operation parameters.""" + + cutoff: datetime | None + cutoff_timestamp: float | None + selected_ids: set[int] | None + audit_trail: bool + internal_job_uuid: str | None + purge_all_logs: bool + + +def resolve_activation_ids( + activation_ids: list[int] | None, + activation_names: list[str] | None, +) -> set[int] | None: + """Validate every selector and return its union, or None for global.""" + if activation_ids is None and activation_names is None: + return None + if activation_ids is not None and ( + not activation_ids + or any( + not isinstance(value, int) or isinstance(value, bool) or value <= 0 + for value in activation_ids + ) + ): + raise ValueError("Activation IDs must be positive integers.") + if activation_names is not None and ( + not activation_names + or any( + not isinstance(value, str) or not value.strip() + for value in activation_names + ) + ): + raise ValueError("Activation names cannot be blank.") + + requested_ids = set(activation_ids or []) + requested_names = set(activation_names or []) + existing_ids = set( + models.Activation.objects.filter(id__in=requested_ids).values_list( + "id", flat=True + ) + ) + if missing := requested_ids - existing_ids: + raise ValueError(f"Unknown activation IDs: {sorted(missing)}.") + + named_activations = models.Activation.objects.filter( + name__in=requested_names + ).values_list("id", "name") + found_names = set() + for activation_id, name in named_activations: + existing_ids.add(activation_id) + found_names.add(name) + if missing := requested_names - found_names: + raise ValueError(f"Unknown activation names: {sorted(missing)}.") + return existing_ids + + +def _marker_text( + cutoff: datetime | None, + purge_time: datetime, + count: int, + tracking_id: str | None, + orphaned: bool, + internal_job_uuid: str | None, +) -> str: + """Format audit marker text with purge metadata.""" + details = [f"Recorded at: {_utc_text(purge_time)}"] + noun = "record" if count == 1 else "records" + if cutoff is None: + text = f"Purged {count} log {noun} globally." + else: + text = f"Purged {count} log {noun} older than {_utc_text(cutoff)}." + if internal_job_uuid is not None: + details.append(f"Purge Job ID: {internal_job_uuid}") + if tracking_id and tracking_id.strip(): + details.append(f"Activation Lifecycle ID: {tracking_id.strip()}") + elif orphaned: + details.append( + "Activation Lifecycle ID: unavailable (orphaned process)" + ) + return f"{text} [{' | '.join(details)}]" + + +def _validate_purge_options( + cutoff: datetime | None, + *, + activation_ids: list[int] | None, + activation_names: list[str] | None, + audit_trail: bool, + internal_job_uuid: str | None, + purge_all_logs: bool, +) -> _PurgeOptions: + """Validate purge parameters and return configuration object.""" + if not isinstance(purge_all_logs, bool): + raise ValueError("purge_all_logs must be a boolean.") + if purge_all_logs: + if any( + selector is not None + for selector in (activation_ids, activation_names) + ): + raise ValueError( + "--purge-all-logs cannot be combined with " + "activation selectors." + ) + if not audit_trail: + raise ValueError("--purge-all-logs requires audit trail.") + selected_ids = None + cutoff_timestamp = cutoff.timestamp() if cutoff is not None else None + else: + if cutoff is None: + raise ValueError("A cutoff is required unless purging all logs.") + selected_ids = resolve_activation_ids(activation_ids, activation_names) + cutoff_timestamp = cutoff.timestamp() + return _PurgeOptions( + cutoff, + cutoff_timestamp, + selected_ids, + audit_trail, + internal_job_uuid, + purge_all_logs, + ) + + +def _write_audit_marker( + process_id: int, + activation_id: int | None, + tracking_id: str | None, + deleted_count: int, + options: _PurgeOptions, +) -> int: + """Create audit marker log record for a purge operation.""" + purge_time = datetime.now(timezone.utc) + marker_timestamp = max( + ceil(purge_time.timestamp()), + ceil(options.cutoff_timestamp) + if options.cutoff_timestamp is not None + else 0, + ) + marker_text = _marker_text( + options.cutoff, + purge_time, + deleted_count, + tracking_id, + activation_id is None, + options.internal_job_uuid, + ) + marker = models.RulebookProcessLog.objects.create( + activation_instance_id=process_id, + log=marker_text, + log_timestamp=marker_timestamp, + log_created_at=purge_time, + ) + return marker.id + + +def _purge_process_logs(process: tuple, options: _PurgeOptions) -> int: + """Delete logs for one process matching purge criteria.""" + process_id, activation_id, tracking_id = process + logs = models.RulebookProcessLog.objects.filter( + activation_instance_id=process_id + ) + if options.cutoff_timestamp is not None: + logs = logs.filter(log_timestamp__lt=options.cutoff_timestamp) + if options.selected_ids is not None: + logs = logs.filter( + activation_instance__activation_id__in=options.selected_ids + ) + with transaction.atomic(): + deleted_count, _ = logs.delete() + if options.audit_trail and deleted_count: + _write_audit_marker( + process_id, + activation_id, + tracking_id, + deleted_count, + options, + ) + return deleted_count + + +def purge_unattached_logs(cutoff_timestamp: float | None = None) -> int: + """Delete logs not linked to any activation instance.""" + logs = models.RulebookProcessLog.objects.filter( + activation_instance_id__isnull=True + ) + if cutoff_timestamp is not None: + logs = logs.filter(log_timestamp__lt=cutoff_timestamp) + deleted_count, _ = logs.delete() + return deleted_count + + +def delete_logs_older_than( + cutoff: datetime | None, + *, + activation_ids: list[int] | None = None, + activation_names: list[str] | None = None, + audit_trail: bool = True, + internal_job_uuid: str | None = None, + purge_all_logs: bool = False, +) -> int: + """Purge qualifying logs one process at a time.""" + options = _validate_purge_options( + cutoff, + activation_ids=activation_ids, + activation_names=activation_names, + audit_trail=audit_trail, + internal_job_uuid=internal_job_uuid, + purge_all_logs=purge_all_logs, + ) + processes = models.RulebookProcess.objects.all() + if options.selected_ids is not None: + processes = processes.filter(activation_id__in=options.selected_ids) + highest_process_id = processes.aggregate(max_id=Max("id"))["max_id"] + if highest_process_id is None: + return 0 + + deleted_count = 0 + last_process_id = 0 + try: + while last_process_id < highest_process_id: + process_batch = list( + processes.filter( + id__gt=last_process_id, id__lte=highest_process_id + ) + .order_by("id") + .values_list( + "id", "activation_id", "activation__log_tracking_id" + )[:PROCESS_PAGE_SIZE] + ) + if not process_batch: + break + for process in process_batch: + deleted_count += _purge_process_logs(process, options) + last_process_id = process_batch[-1][0] + except Exception as exc: + raise LogPurgeError( + f"Log purge failed after {deleted_count} committed rows: {exc}", + deleted_count, + ) from exc + return deleted_count diff --git a/src/aap_eda/core/utils/log_retention.py b/src/aap_eda/core/utils/log_retention.py new file mode 100644 index 000000000..e49a4bd17 --- /dev/null +++ b/src/aap_eda/core/utils/log_retention.py @@ -0,0 +1,44 @@ +# Copyright 2026 Red Hat, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from datetime import datetime, timezone as datetime_timezone + +import isodate + + +def retention_cutoff( + duration: str, *, now: datetime | None = None +) -> datetime: + """Return the UTC cutoff for a positive ISO 8601 duration. + + Calendar months and years use ``isodate.Duration`` subtraction semantics: + the month is shifted and the day is clamped to that month's last day. + Other duration components are subtracted as elapsed time. + """ + now = now or datetime.now(datetime_timezone.utc) + if now.tzinfo is None or now.utcoffset() is None: + raise ValueError("The current time must be timezone-aware.") + now = now.astimezone(datetime_timezone.utc) + + try: + parsed = isodate.parse_duration(duration) + cutoff = now - parsed + except (TypeError, ValueError, OverflowError) as exc: + raise ValueError( + "retention_duration must be a valid ISO 8601 duration." + ) from exc + + if cutoff >= now: + raise ValueError("retention_duration must produce a past cutoff.") + return cutoff diff --git a/src/aap_eda/tasks/management.py b/src/aap_eda/tasks/management.py new file mode 100644 index 000000000..1c54c3451 --- /dev/null +++ b/src/aap_eda/tasks/management.py @@ -0,0 +1,155 @@ +# Copyright 2026 Red Hat, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import logging +import re +from uuid import uuid4 + +from ansible_base.lib.utils.db import advisory_lock +from dispatcherd.publish import submit_task +from django.conf import settings +from django.core.management import call_command + +from aap_eda import utils +from aap_eda.core.utils.delete_log_util import LogPurgeError + +logger = logging.getLogger(__name__) + +MANAGEMENT_TASKS_QUEUE = "default" +LOG_PURGE_TASK_TIMEOUT_SECONDS = 3600 +_DELETED_COUNT = re.compile(r"^Purged (\d+) log records?") + + +def submit_log_purge( + *, + before_date: str | None = None, + retention_duration: str | None = None, + audit_trail: bool = True, + activation_ids: list[int] | None = None, + purge_all_logs: bool = False, +) -> str: + """Submit a fixed-purpose log purge task.""" + command_options = {"audit_trail": audit_trail} + if before_date is not None: + command_options["date"] = before_date + if retention_duration is not None: + command_options["retention_duration"] = retention_duration + if activation_ids is not None: + command_options["activation-ids"] = activation_ids + if purge_all_logs: + command_options["purge-all-logs"] = True + + return aap_eda_manage_commands( + "purge_log_records", + timeout=LOG_PURGE_TASK_TIMEOUT_SECONDS, + **command_options, + ) + + +def aap_eda_manage_commands( + command: str, *, timeout: int | None = None, **command_options +) -> str: + """Submit a Django management command to the default worker queue. + + The API uses ``submit_log_purge`` so its request cannot choose a command. + This generic entry point remains available to internal callers. + """ + job_id = str(uuid4()) + if command == "purge_log_records": + command_options["internal-job-uuid"] = job_id + job_data, _ = submit_task( + _aap_eda_manage, + args=(job_id, command, command_options), + queue=utils.sanitize_postgres_identifier(MANAGEMENT_TASKS_QUEUE), + timeout=( + settings.DISPATCHERD_PROJECT_TASK_TIMEOUT + if timeout is None + else timeout + ), + uuid=job_id, + ) + return str(job_data["uuid"]) + + +def _aap_eda_manage(job_id: str, command: str, command_options: dict) -> dict: + """Run a management command under an advisory lock.""" + logger.info( + "Management command started job_id=%s command=%s", job_id, command + ) + with advisory_lock(command, wait=False) as acquired: + if not acquired: + reason = f"another task holds the advisory lock for {command}" + logger.warning( + "Management command skipped job_id=%s command=%s " + "status=skipped reason=%s", + job_id, + command, + reason, + ) + return { + "job_id": job_id, + "command": command, + "status": "skipped", + "reason": reason, + } + + try: + command_output = call_command( + command, no_color=True, **command_options + ) + except Exception as exc: + cause = exc.__cause__ + if command == "purge_log_records" and isinstance( + cause, LogPurgeError + ): + logger.exception( + "Management command failed job_id=%s command=%s " + "status=failed deleted_count=%d", + job_id, + command, + cause.deleted_count, + ) + else: + logger.exception( + "Management command failed job_id=%s command=%s " + "status=failed", + job_id, + command, + ) + raise + + result = { + "job_id": job_id, + "command": command, + "status": "completed", + } + if command == "purge_log_records": + match = _DELETED_COUNT.match(command_output or "") + deleted_count = int(match.group(1)) if match else 0 + result["deleted"] = deleted_count + result["status"] = "completed" if deleted_count else "no-op" + logger.info( + "Log purge finished job_id=%s status=%s deleted_count=%d", + job_id, + result["status"], + deleted_count, + ) + else: + logger.info( + "Management command finished job_id=%s command=%s " + "status=completed", + job_id, + command, + ) + return result diff --git a/tests/integration/api/test_activation_instance.py b/tests/integration/api/test_activation_instance.py index 746ead842..ca909fb9d 100644 --- a/tests/integration/api/test_activation_instance.py +++ b/tests/integration/api/test_activation_instance.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +from datetime import datetime, timezone from typing import Any, Dict, List import pytest @@ -19,6 +20,7 @@ from rest_framework.test import APIClient from aap_eda.core import enums, models +from aap_eda.core.utils.delete_log_util import delete_logs_older_than from aap_eda.utils.log_sanitizer import REDACTED_STRING from tests.integration.constants import api_url_v1 @@ -117,6 +119,44 @@ def test_list_logs_from_activation_instance( ] +@pytest.mark.django_db +def test_list_logs_returns_actual_purge_marker_time( + default_activation_instances: List[models.RulebookProcess], + admin_client: APIClient, +): + instance = default_activation_instances[0] + models.RulebookProcessLog.objects.create( + activation_instance=instance, + log="old log", + log_timestamp=1, + ) + before_purge = datetime.now(timezone.utc) + assert ( + delete_logs_older_than( + datetime(2024, 1, 1, tzinfo=timezone.utc), + activation_ids=[instance.activation_id], + ) + == 1 + ) + after_purge = datetime.now(timezone.utc) + marker = models.RulebookProcessLog.objects.get( + activation_instance=instance, log__startswith="Purged " + ) + + response = admin_client.get( + f"{api_url_v1}/activation-instances/{instance.id}/logs/" + ) + + assert response.status_code == status.HTTP_200_OK + marker_data = next( + entry for entry in response.data["results"] if entry["id"] == marker.id + ) + assert before_purge <= marker.log_created_at <= after_purge + assert marker_data[ + "log_created_at" + ] == marker.log_created_at.isoformat().replace("+00:00", "Z") + + @pytest.mark.django_db def test_list_activation_instance_logs_filter( default_activation_instances: List[models.RulebookProcess], diff --git a/tests/integration/api/test_log_purge.py b/tests/integration/api/test_log_purge.py new file mode 100644 index 000000000..8c6ec0466 --- /dev/null +++ b/tests/integration/api/test_log_purge.py @@ -0,0 +1,293 @@ +# Copyright 2026 Red Hat, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 + +from unittest import mock +from uuid import UUID + +import pytest +from rest_framework import status + +from aap_eda.tasks.management import LOG_PURGE_TASK_TIMEOUT_SECONDS +from tests.integration.constants import api_url_v1 + +PURGE_URL = f"{api_url_v1}/logs/purge/" +JOB_ID = UUID("c3dd8dd8-b1d5-49e0-86e9-216cb2c942a8") + + +@pytest.mark.django_db +def test_superuser_queues_scoped_purge_with_one_eda_job_id( + superuser_client, default_activation, new_activation +): + with mock.patch( + "aap_eda.tasks.management.uuid4", return_value=JOB_ID + ), mock.patch( + "aap_eda.tasks.management.submit_task", + return_value=({"uuid": str(JOB_ID)}, "default"), + ) as submit_task: + response = superuser_client.post( + PURGE_URL, + { + "retention_duration": "P30D", + "activation_ids": [default_activation.id], + "activation_names": [new_activation.name], + }, + format="json", + ) + + assert response.status_code == status.HTTP_202_ACCEPTED + assert response.data == {"job_id": str(JOB_ID)} + submit_task.assert_called_once() + (task,) = submit_task.call_args.args + options = submit_task.call_args.kwargs + assert task.__name__ == "_aap_eda_manage" + assert options["queue"] == "default" + assert options["timeout"] == LOG_PURGE_TASK_TIMEOUT_SECONDS + assert options["uuid"] == str(JOB_ID) + assert options["args"] == ( + str(JOB_ID), + "purge_log_records", + { + "retention_duration": "P30D", + "audit_trail": True, + "activation-ids": sorted( + [default_activation.id, new_activation.id] + ), + "internal-job-uuid": str(JOB_ID), + }, + ) + + +@pytest.mark.django_db +def test_name_only_scope_is_queued_as_activation_id( + superuser_client, default_activation +): + with mock.patch( + "aap_eda.tasks.management.submit_task", + return_value=({"uuid": str(JOB_ID)}, "default"), + ) as submit_task: + response = superuser_client.post( + PURGE_URL, + { + "before_date": "2020-01-01", + "activation_names": [default_activation.name], + }, + format="json", + ) + + assert response.status_code == status.HTTP_202_ACCEPTED + options = submit_task.call_args.kwargs["args"][2] + assert options["activation-ids"] == [default_activation.id] + assert "activation-names" not in options + + +@pytest.mark.django_db +def test_superuser_queues_global_purge_with_absolute_date(superuser_client): + with mock.patch( + "aap_eda.tasks.management.uuid4", return_value=JOB_ID + ), mock.patch( + "aap_eda.tasks.management.submit_task", + return_value=({"uuid": str(JOB_ID)}, "default"), + ) as submit_task: + response = superuser_client.post( + PURGE_URL, {"before_date": "2020-01-01"}, format="json" + ) + + assert response.status_code == status.HTTP_202_ACCEPTED + assert response.data == {"job_id": str(JOB_ID)} + assert submit_task.call_args.kwargs["args"][2] == { + "date": "2020-01-01", + "audit_trail": True, + "internal-job-uuid": str(JOB_ID), + } + + +@pytest.mark.django_db +def test_superuser_queues_purge_all_logs_without_cutoff(superuser_client): + with mock.patch( + "aap_eda.tasks.management.uuid4", return_value=JOB_ID + ), mock.patch( + "aap_eda.tasks.management.submit_task", + return_value=({"uuid": str(JOB_ID)}, "default"), + ) as submit_task: + response = superuser_client.post( + PURGE_URL, {"purge-all-logs": True}, format="json" + ) + + assert response.status_code == status.HTTP_202_ACCEPTED + assert response.data == {"job_id": str(JOB_ID)} + assert submit_task.call_args.kwargs["args"] == ( + str(JOB_ID), + "purge_log_records", + { + "audit_trail": True, + "purge-all-logs": True, + "internal-job-uuid": str(JOB_ID), + }, + ) + + +@pytest.mark.django_db +def test_superuser_queues_purge_all_logs_with_cutoff(superuser_client): + with mock.patch( + "aap_eda.tasks.management.submit_task", + return_value=({"uuid": str(JOB_ID)}, "default"), + ) as submit_task: + response = superuser_client.post( + PURGE_URL, + {"purge-all-logs": True, "before_date": "2020-01-01"}, + format="json", + ) + + assert response.status_code == status.HTTP_202_ACCEPTED + command_options = submit_task.call_args.kwargs["args"][2] + assert command_options["purge-all-logs"] is True + assert command_options["date"] == "2020-01-01" + + +@pytest.mark.django_db +def test_superuser_queues_purge_all_logs_with_retention_duration( + superuser_client, +): + with mock.patch( + "aap_eda.tasks.management.submit_task", + return_value=({"uuid": str(JOB_ID)}, "default"), + ) as submit_task: + response = superuser_client.post( + PURGE_URL, + {"purge-all-logs": True, "retention_duration": "P30D"}, + format="json", + ) + + assert response.status_code == status.HTTP_202_ACCEPTED + command_options = submit_task.call_args.kwargs["args"][2] + assert command_options["purge-all-logs"] is True + assert command_options["retention_duration"] == "P30D" + + +@pytest.mark.django_db +def test_api_keeps_activation_scope_when_audit_is_disabled( + superuser_client, default_activation +): + with mock.patch( + "aap_eda.tasks.management.uuid4", return_value=JOB_ID + ), mock.patch( + "aap_eda.tasks.management.submit_task", + return_value=({"uuid": str(JOB_ID)}, "default"), + ) as submit_task: + response = superuser_client.post( + PURGE_URL, + { + "before_date": "2020-01-01", + "activation_ids": [default_activation.id], + "audit_trail": False, + }, + format="json", + ) + + assert response.status_code == status.HTTP_202_ACCEPTED + assert submit_task.call_args.kwargs["args"][2] == { + "date": "2020-01-01", + "audit_trail": False, + "activation-ids": [default_activation.id], + "internal-job-uuid": str(JOB_ID), + } + + +@pytest.mark.django_db +def test_purge_requires_superuser(admin_client): + response = admin_client.post(PURGE_URL, {}, format="json") + assert response.status_code == status.HTTP_403_FORBIDDEN + + +@pytest.mark.django_db +def test_anonymous_purge_uses_api_permission_denial(base_client): + response = base_client.post(PURGE_URL, {}, format="json") + assert response.status_code == status.HTTP_401_UNAUTHORIZED + + +@pytest.mark.django_db +@pytest.mark.parametrize( + "payload", + [ + {}, + {"before_date": "2999-01-01"}, + {"before_date": "not-a-date"}, + {"before_date": "2020-01-01", "retention_duration": "P30D"}, + {"retention_duration": "not-a-duration"}, + {"retention_duration": "P0D"}, + {"retention_duration": "-P1D"}, + {"purge-all-logs": False}, + {"purge_all_logs": True}, + { + "purge-all-logs": True, + "before_date": "2020-01-01", + "retention_duration": "P30D", + }, + {"purge-all-logs": True, "before_date": "2999-01-01"}, + {"purge-all-logs": True, "retention_duration": "P0D"}, + {"purge-all-logs": True, "audit_trail": False}, + {"purge-all-logs": True, "activation_ids": [1]}, + {"purge-all-logs": True, "activation_names": ["example"]}, + {"before_date": "2020-01-01", "activation_ids": [0]}, + {"before_date": "2020-01-01", "activation_ids": []}, + {"before_date": "2020-01-01", "activation_names": [""]}, + {"before_date": "2020-01-01", "activation_names": [" "]}, + {"before_date": "2020-01-01", "activation_ids": [99999999]}, + {"before_date": "2020-01-01", "activation_id": 1}, + {"before_date": "2020-01-01", "activation_id": "invalid"}, + {"before_date": "2020-01-01", "activation_id": None}, + {"before_date": "2020-01-01", "activation_namess": ["demo"]}, + { + "before_date": "2020-01-01", + "activation_names": ["missing-activation"], + }, + ], +) +def test_invalid_purge_request_is_rejected_before_enqueue( + superuser_client, payload +): + with mock.patch("aap_eda.tasks.management.submit_task") as submit_task: + response = superuser_client.post(PURGE_URL, payload, format="json") + + assert response.status_code == status.HTTP_400_BAD_REQUEST + submit_task.assert_not_called() + + +@pytest.mark.django_db +@pytest.mark.parametrize("invalid_kind", ["id", "name"]) +def test_mixed_valid_and_unknown_scope_never_queues( + superuser_client, default_activation, invalid_kind +): + payload = { + "before_date": "2020-01-01", + "activation_ids": [default_activation.id], + } + if invalid_kind == "id": + payload["activation_ids"].append(default_activation.id + 99999999) + else: + payload["activation_names"] = ["missing-activation"] + + with mock.patch("aap_eda.tasks.management.submit_task") as submit_task: + response = superuser_client.post(PURGE_URL, payload, format="json") + + assert response.status_code == status.HTTP_400_BAD_REQUEST + submit_task.assert_not_called() + + +@pytest.mark.django_db +def test_enqueue_failure_is_not_reported_as_accepted(superuser_client): + with mock.patch( + "aap_eda.tasks.management.submit_task", + side_effect=RuntimeError("queue unavailable"), + ): + response = superuser_client.post( + PURGE_URL, {"before_date": "2020-01-01"}, format="json" + ) + + assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR + assert response.data == {"detail": "Unable to queue log purge job."} diff --git a/tests/integration/api/test_openapi.py b/tests/integration/api/test_openapi.py index 2250d8880..e86c74e0a 100644 --- a/tests/integration/api/test_openapi.py +++ b/tests/integration/api/test_openapi.py @@ -87,3 +87,43 @@ def test_v1_openapi_credential_type_inputs_schema(admin_client): assert "properties" not in request_inputs assert request_inputs["additionalProperties"] is True assert request_inputs["example"]["required"] == ["username"] + + +@pytest.mark.django_db +def test_log_purge_openapi_contract(admin_client): + response = admin_client.get(f"{api_url_v1}/openapi.json") + assert response.status_code == status.HTTP_200_OK + + spec = response.json() + purge_path = next( + path for path in spec["paths"] if path.endswith("/logs/purge/") + ) + operation = spec["paths"][purge_path]["post"] + request_ref = operation["requestBody"]["content"]["application/json"][ + "schema" + ]["$ref"] + request_schema = spec["components"]["schemas"][ + request_ref.rsplit("/", 1)[1] + ] + request_properties = request_schema["properties"] + assert set(request_properties) == { + "before_date", + "retention_duration", + "audit_trail", + "activation_ids", + "activation_names", + "purge-all-logs", + } + assert request_properties["audit_trail"]["default"] is True + assert request_properties["purge-all-logs"]["default"] is False + assert "internal_job_uuid" not in request_properties + + responses = operation["responses"] + assert {"202", "400", "403"}.issubset(responses) + response_ref = responses["202"]["content"]["application/json"]["schema"][ + "$ref" + ] + response_schema = spec["components"]["schemas"][ + response_ref.rsplit("/", 1)[1] + ] + assert set(response_schema["properties"]) == {"job_id"} diff --git a/tests/integration/api/test_root.py b/tests/integration/api/test_root.py index 416fe1e73..99ee27419 100644 --- a/tests/integration/api/test_root.py +++ b/tests/integration/api/test_root.py @@ -35,6 +35,7 @@ "/projects/", "/rulebooks/", "/activations/", + "/logs/", "/activation-instances/", "/audit-rules/", "/users/", @@ -91,6 +92,7 @@ "/projects/", "/rulebooks/", "/activations/", + "/logs/", "/activation-instances/", "/audit-rules/", "/users/", diff --git a/tests/integration/commands/test_purge_log_records.py b/tests/integration/commands/test_purge_log_records.py index 376aaa69e..84097b7bf 100644 --- a/tests/integration/commands/test_purge_log_records.py +++ b/tests/integration/commands/test_purge_log_records.py @@ -12,15 +12,25 @@ # See the License for the specific language governing permissions and # limitations under the License. -from datetime import timedelta +from datetime import datetime, timedelta, timezone as datetime_timezone +from math import ceil +from unittest import mock +from uuid import uuid4 import pytest from django.core.management import call_command from django.core.management.base import CommandError from django.db import connection, transaction +from django.db.models.query import QuerySet from django.utils import timezone from aap_eda.core import enums, models +from aap_eda.core.utils import delete_log_util +from aap_eda.core.utils.log_retention import retention_cutoff + + +def _audit_markers(): + return models.RulebookProcessLog.objects.filter(log__startswith="Purged ") @pytest.fixture @@ -158,36 +168,32 @@ def prepare_log_records( @pytest.mark.django_db -def test_purge_log_records_missing_required_params(): - with pytest.raises( - CommandError, - match="Error: the following arguments are required: --date", - ): +def test_purge_log_records_requires_explicit_cutoff(): + with pytest.raises(CommandError, match="Exactly one"): call_command("purge_log_records") @pytest.mark.django_db -def test_purge_log_records_with_nonexist_activation(capsys): - # With global purging, even non-existent activation filters will - # still purge old logs globally (if any exist) - args = ("--activation-ids", "42", "--date", "2024-10-01") - call_command("purge_log_records", args) - captured = capsys.readouterr() - - # Should report either no old logs found, or global purge count - assert ( - "No log records found older than 2024-10-01" in captured.out - or "Purged" in captured.out +@pytest.mark.parametrize( + "selector, expected", + [ + (("--activation-ids", "999999999"), "Unknown activation IDs"), + (("--activation-names", "missing-name"), "Unknown activation names"), + ], +) +def test_unknown_selector_rejects_before_deleting( + prepare_log_records, selector, expected +): + original_ids = set( + models.RulebookProcessLog.objects.values_list("id", flat=True) ) - args = ("--activation-names", "na", "--date", "2024-10-01") - call_command("purge_log_records", args) - captured = capsys.readouterr() + with pytest.raises(CommandError, match=expected): + call_command("purge_log_records", *selector, "--date", "2024-10-01") - assert ( - "No log records found older than 2024-10-01" in captured.out - or "Purged" in captured.out - ) + assert set( + models.RulebookProcessLog.objects.values_list("id", flat=True) + ) == (original_ids) @pytest.mark.parametrize( @@ -210,8 +216,8 @@ def test_purge_log_records( if identifiers == "ids": args = ( "--activation-ids", - activations[0].id, - activations[1].id, + str(activations[0].id), + str(activations[1].id), "--date", date_str, ) @@ -229,39 +235,39 @@ def test_purge_log_records( date_str, ) - call_command(command, args) + call_command(command, *args) captured = capsys.readouterr() if cutoff_days < 10: - # Global purge: all original logs deleted, audit logs created for - # all 4 RulebookProcess instances + # All four processes lose rows and receive their own marker. assert models.RulebookProcessLog.objects.count() == 4 for log_record in models.RulebookProcessLog.objects.all(): - assert ( - f"All log records older than {date_str} were purged at" - in log_record.log + assert f"Purged 2 log records older than {date_str}" in ( + log_record.log ) + assert "[Recorded at:" in log_record.log else: - # Global purge: only 30-day logs deleted, 10-day logs remain, - # audit logs created for all 4 RulebookProcess instances - assert models.RulebookProcessLog.objects.count() == 8 + # Only the two older processes lose rows and receive markers. + assert models.RulebookProcessLog.objects.count() == 6 - # Check audit logs exist for all instances audit_logs = models.RulebookProcessLog.objects.filter( - log__contains=f"All log records older than {date_str} were purged" + log__contains=f"Purged 2 log records older than {date_str}" ) - assert audit_logs.count() == 4 + assert audit_logs.count() == 2 + assert all("[Recorded at:" in row.log for row in audit_logs) # Check that 10-day logs still exist (not purged) - original_logs = models.RulebookProcessLog.objects.exclude( - log__contains="purged" + original_logs = models.RulebookProcessLog.objects.filter( + log__startswith="activation-instance-10-days-ago" ) assert original_logs.count() == 4 assert "Purged" in captured.out - assert "globally" in captured.out + assert ( + "globally" if identifiers == "none" else "for selected activations" + ) in captured.out @pytest.fixture @@ -361,151 +367,1040 @@ def null_activation_test_data( @pytest.mark.django_db -def test_purge_without_activation_filter_handles_null_correctly( - null_activation_test_data, capsys +def test_global_purge_includes_orphan_and_marks_each_process( + null_activation_test_data, ): - """Test purge without activation filter includes NULL records. - This tests the customer scenario.""" + cutoff_date = (timezone.now() - timedelta(days=10)).strftime("%Y-%m-%d") + output = call_command("purge_log_records", "--date", cutoff_date) - # Get initial counts - initial_log_count = models.RulebookProcessLog.objects.filter( + assert output.startswith("Purged 4 log records") + assert "globally" in output + assert not models.RulebookProcessLog.objects.filter( log__contains="for null test" - ).count() - assert initial_log_count == 4, "Should have 4 test log records" + ).exists() + markers = _audit_markers() + assert markers.count() == 2 + orphan_marker = markers.get( + activation_instance=null_activation_test_data["null_process"] + ) + assert "Purged 2 log records older than" in orphan_marker.log + assert "Activation Lifecycle ID: unavailable (orphaned process)" in ( + orphan_marker.log + ) - # Test purge without activation filtering (customer's command scenario) + +@pytest.mark.django_db +@pytest.mark.parametrize("selector", ["id", "name"]) +def test_scoped_purge_preserves_orphan(null_activation_test_data, selector): cutoff_date = (timezone.now() - timedelta(days=10)).strftime("%Y-%m-%d") + activation = null_activation_test_data["activation"] + option, value = ( + ("--activation-ids", str(activation.id)) + if selector == "id" + else ("--activation-names", activation.name) + ) - call_command("purge_log_records", "--date", cutoff_date, verbosity=3) + output = call_command( + "purge_log_records", option, value, "--date", cutoff_date + ) - captured = capsys.readouterr() + assert output.startswith("Purged 2 log records") + assert "for selected activations" in output + assert ( + models.RulebookProcessLog.objects.filter( + activation_instance=null_activation_test_data["null_process"], + log__contains="for null test", + ).count() + == 2 + ) + assert not models.RulebookProcessLog.objects.filter( + activation_instance=null_activation_test_data["normal_process"], + log__contains="for null test", + ).exists() - # Check that logs were purged - remaining_test_logs = models.RulebookProcessLog.objects.filter( - log__contains="for null test" - ).exclude(log__contains="purged") - # Both NULL and normal activation logs should be purged in this scenario - assert ( - remaining_test_logs.count() == 0 - ), "All test logs should be purged when no activation filter is used" - assert "Purged" in captured.out - assert "globally" in captured.out +@pytest.mark.django_db +def test_purge_without_audit_trail_does_not_create_audit_logs( + prepare_log_records, capsys +): + cutoff = (timezone.now() - timedelta(days=15)).strftime("%Y-%m-%d") + output = call_command( + "purge_log_records", "--date", cutoff, "--audit-trail=false" + ) + + assert output.startswith("Purged 4 log records") + assert models.RulebookProcessLog.objects.count() == 4 + assert not _audit_markers().exists() + assert "Purged 4 log records" in capsys.readouterr().out @pytest.mark.django_db -def test_purge_with_activation_filter_now_includes_null_fix( - null_activation_test_data, capsys +@pytest.mark.parametrize( + "args", + [ + ("--date", "2026-01-01", "--retention-duration", "P1D"), + ("--retention-duration", "P0D"), + ("--retention-duration", "-P1D"), + ("--retention-duration", "not-a-duration"), + ("--date", "not-a-date"), + ("--date", "2999-01-01"), + ], +) +def test_purge_rejects_invalid_retention_arguments(args): + with pytest.raises(CommandError): + call_command("purge_log_records", *args) + + +@pytest.mark.parametrize( + "now, duration, expected", + [ + ( + datetime(2026, 3, 31, 12, tzinfo=datetime_timezone.utc), + "P1M", + datetime(2026, 2, 28, 12, tzinfo=datetime_timezone.utc), + ), + ( + datetime(2024, 3, 31, 12, tzinfo=datetime_timezone.utc), + "P1M", + datetime(2024, 2, 29, 12, tzinfo=datetime_timezone.utc), + ), + ( + datetime(2024, 2, 29, 12, tzinfo=datetime_timezone.utc), + "P1Y", + datetime(2023, 2, 28, 12, tzinfo=datetime_timezone.utc), + ), + ( + datetime( + 2026, + 10, + 1, + 0, + 30, + tzinfo=timezone.get_fixed_timezone(120), + ), + "P1D", + datetime(2026, 9, 29, 22, 30, tzinfo=datetime_timezone.utc), + ), + ( + datetime(2026, 10, 21, 12, tzinfo=datetime_timezone.utc), + "PT30S", + datetime(2026, 10, 21, 11, 59, 30, tzinfo=datetime_timezone.utc), + ), + ( + datetime(2026, 10, 21, 12, tzinfo=datetime_timezone.utc), + "PT5M", + datetime(2026, 10, 21, 11, 55, tzinfo=datetime_timezone.utc), + ), + ( + datetime(2026, 10, 21, 12, tzinfo=datetime_timezone.utc), + "PT2H", + datetime(2026, 10, 21, 10, tzinfo=datetime_timezone.utc), + ), + ( + datetime(2026, 10, 21, 12, tzinfo=datetime_timezone.utc), + "P2W", + datetime(2026, 10, 7, 12, tzinfo=datetime_timezone.utc), + ), + ], +) +def test_retention_cutoff_calendar_and_timezone_policy( + now, duration, expected ): - """Purge includes NULL records with filters.""" + assert retention_cutoff(duration, now=now) == expected - # Get initial counts - initial_log_count = models.RulebookProcessLog.objects.filter( - log__contains="for null test" - ).count() - assert initial_log_count == 4, "Should have 4 test log records" - # Test purge WITH activation filtering - this should now include - # NULL records - cutoff_date = (timezone.now() - timedelta(days=10)).strftime("%Y-%m-%d") - activation_id = null_activation_test_data["activation"].id +@pytest.mark.parametrize("filter_by", ["ids", "names"]) +@pytest.mark.django_db +def test_activation_filter_limits_deletion_and_markers( + prepare_log_records, filter_by +): + selected_activation = prepare_log_records[0] + cutoff = (timezone.now() - timedelta(days=5)).strftime("%Y-%m-%d") + option = "--activation-ids" if filter_by == "ids" else "--activation-names" + value = ( + str(selected_activation.id) + if filter_by == "ids" + else selected_activation.name + ) - call_command( + output = call_command( "purge_log_records", - "--activation-ids", - str(activation_id), + option, + value, "--date", - cutoff_date, - verbosity=3, + cutoff, ) - captured = capsys.readouterr() + audit_logs = _audit_markers() + assert output.startswith("Purged 4 log records") + assert audit_logs.count() == 2 + assert set( + audit_logs.values_list("activation_instance__activation_id", flat=True) + ) == {selected_activation.id} + cutoff_timestamp = ( + datetime.strptime(cutoff, "%Y-%m-%d") + .replace(tzinfo=datetime_timezone.utc) + .timestamp() + ) + assert all( + log_timestamp >= cutoff_timestamp + for log_timestamp in audit_logs.values_list("log_timestamp", flat=True) + ) + assert ( + models.RulebookProcessLog.objects.filter( + activation_instance__activation_id=prepare_log_records[1].id, + log__startswith="activation-instance-", + ).count() + == 4 + ) - # Check what logs remain - remaining_test_logs = models.RulebookProcessLog.objects.filter( - log__contains="for null test" - ).exclude(log__contains="purged") - # Verify the fix: NULL activation logs should now be purged - null_activation_logs = [ - log - for log in remaining_test_logs - if log.activation_instance.activation_id is None +@pytest.mark.django_db +def test_retention_duration_deletes_strictly_older_rows_across_activations( + default_activation_instances, +): + cutoff = datetime(2026, 1, 1, tzinfo=datetime_timezone.utc) + older, equal, newer = [ + models.RulebookProcessLog.objects.create( + log=f"boundary-{timestamp}", + activation_instance=default_activation_instances[index % 2], + log_timestamp=timestamp, + ) + for index, timestamp in enumerate( + ( + cutoff.timestamp() - 1, + cutoff.timestamp(), + cutoff.timestamp() + 1, + ) + ) ] - # This validates the fix: NULL activation logs should now be - # included in purge - assert len(null_activation_logs) == 0, ( - "All NULL activation logs should be purged " - "when using --activation-ids filter" + deleted = delete_log_util.delete_logs_older_than(cutoff) + + assert deleted == 1 + assert not models.RulebookProcessLog.objects.filter(pk=older.pk).exists() + assert ( + models.RulebookProcessLog.objects.filter( + pk__in=[equal.pk, newer.pk] + ).count() + == 2 ) - # Also verify that normal activation logs WERE purged - normal_activation_logs = [ - log - for log in remaining_test_logs - if log.activation_instance.activation_id is not None - ] + +@pytest.mark.django_db +def test_retention_deletes_qualifying_rows_in_one_call_per_process( + default_activation_instance, monkeypatch +): + logs = models.RulebookProcessLog.objects.bulk_create( + [ + models.RulebookProcessLog( + log=f"old-{index}", + activation_instance=default_activation_instance, + log_timestamp=1, + ) + for index in range(5) + ] + ) + original_delete = QuerySet.delete + deleted_query_sizes = [] + + def record_delete(queryset): + deleted_query_sizes.append(queryset.count()) + return original_delete(queryset) + + monkeypatch.setattr(QuerySet, "delete", record_delete) + + assert ( + delete_log_util.delete_logs_older_than( + datetime(1970, 1, 1, 0, 0, 2, tzinfo=datetime_timezone.utc), + ) + == 5 + ) + assert not models.RulebookProcessLog.objects.filter( + pk__in=[log.pk for log in logs] + ).exists() + assert deleted_query_sizes == [5] + marker = _audit_markers().get() + assert "Purged 5 log records older than" in marker.log + assert marker.log_created_at is not None assert ( - len(normal_activation_logs) == 0 - ), "Normal activation logs should have been purged" + f"Recorded at: {delete_log_util._utc_text(marker.log_created_at)}" + in marker.log + ) + - # Verify that the command output mentions orphaned records - assert "Audit trail will include" in captured.out +@pytest.mark.django_db +def test_retention_does_not_revisit_process_after_deletion( + default_activation_instance, monkeypatch +): + models.RulebookProcessLog.objects.bulk_create( + [ + models.RulebookProcessLog( + log=f"snapshot-{index}", + activation_instance=default_activation_instance, + log_timestamp=1, + ) + for index in range(2) + ] + ) + original_delete = QuerySet.delete + new_log = [] + calls = 0 + + def insert_after_delete(queryset): + nonlocal calls + result = original_delete(queryset) + calls += 1 + if calls == 1: + new_log.append( + models.RulebookProcessLog.objects.create( + log="arrived-during-purge", + activation_instance=default_activation_instance, + log_timestamp=1, + ) + ) + return result + + monkeypatch.setattr(QuerySet, "delete", insert_after_delete) + deleted = delete_log_util.delete_logs_older_than( + datetime(1970, 1, 1, 0, 0, 2, tzinfo=datetime_timezone.utc), + ) + + assert deleted == 2 + assert models.RulebookProcessLog.objects.filter(pk=new_log[0].pk).exists() + + +@pytest.mark.django_db +def test_delete_rechecks_cutoff_when_executed( + default_activation_instance, monkeypatch +): + log = models.RulebookProcessLog.objects.create( + log="changed-between-selection-and-delete", + activation_instance=default_activation_instance, + log_timestamp=1, + ) + original_delete = QuerySet.delete + + def move_log_past_cutoff_before_delete(queryset): + models.RulebookProcessLog.objects.filter(pk=log.pk).update( + log_timestamp=3 + ) + return original_delete(queryset) + + monkeypatch.setattr(QuerySet, "delete", move_log_past_cutoff_before_delete) + cutoff = datetime(1970, 1, 1, 0, 0, 2, tzinfo=datetime_timezone.utc) + + assert delete_log_util.delete_logs_older_than(cutoff) == 0 + log.refresh_from_db() + assert log.log_timestamp == 3 + assert models.RulebookProcessLog.objects.count() == 1 + + +@pytest.mark.django_db +def test_retention_with_no_matching_rows_returns_zero( + default_activation_instance, +): + models.RulebookProcessLog.objects.create( + log="new", + activation_instance=default_activation_instance, + log_timestamp=2, + ) + assert ( + delete_log_util.delete_logs_older_than( + datetime(1970, 1, 1, 0, 0, 2, tzinfo=datetime_timezone.utc) + ) + == 0 + ) + + +@pytest.mark.django_db(transaction=True) +def test_retention_failure_reports_committed_process_count( + default_activation_instances, monkeypatch +): + models.RulebookProcessLog.objects.bulk_create( + [ + models.RulebookProcessLog( + log=f"partial-{index}", + activation_instance=process, + log_timestamp=1, + ) + for index, process in enumerate(default_activation_instances) + ] + ) + original_delete = QuerySet.delete + calls = 0 + + def fail_second_delete(queryset): + nonlocal calls + calls += 1 + if calls == 2: + raise RuntimeError("injected process failure") + return original_delete(queryset) + + monkeypatch.setattr(QuerySet, "delete", fail_second_delete) + cutoff = datetime(1970, 1, 1, 0, 0, 2, tzinfo=datetime_timezone.utc) + with pytest.raises(delete_log_util.LogPurgeError) as error: + delete_log_util.delete_logs_older_than(cutoff) + + assert error.value.deleted_count == 1 + assert models.RulebookProcessLog.objects.count() == 2 + marker = _audit_markers().get() + assert "Purged 1 log record older than" in marker.log assert ( - "orphaned" in captured.out - ), "Should mention orphaned records in output" + models.RulebookProcessLog.objects.filter( + log__startswith="partial-" + ).count() + == 1 + ) @pytest.mark.django_db -def test_purge_with_activation_name_filter_also_includes_null_fix( - null_activation_test_data, capsys +def test_retention_duration_cli_uses_execution_time_cutoff( + default_activation_instance, ): - """Test that the fix also works when using --activation-names parameter.""" + cutoff_epoch = datetime( + 2026, 10, 20, tzinfo=datetime_timezone.utc + ).timestamp() + logs = models.RulebookProcessLog.objects.bulk_create( + [ + models.RulebookProcessLog( + log=f"duration-{index}", + activation_instance=default_activation_instance, + log_timestamp=timestamp, + ) + for index, timestamp in enumerate( + (cutoff_epoch - 1, cutoff_epoch, cutoff_epoch + 1) + ) + ] + ) - # Get initial counts - initial_log_count = models.RulebookProcessLog.objects.filter( - log__contains="for null test" - ).count() - assert initial_log_count == 4, "Should have 4 test log records" + with mock.patch( + "aap_eda.core.utils.log_retention.datetime" + ) as mocked_datetime: + mocked_datetime.now.return_value = datetime( + 2026, 10, 21, tzinfo=datetime_timezone.utc + ) + mocked_datetime.timezone = datetime_timezone + output = call_command( + "purge_log_records", "--retention-duration", "P1D" + ) - # Test purge with activation NAME filtering - should now include - # NULL records (fix) - cutoff_date = (timezone.now() - timedelta(days=10)).strftime("%Y-%m-%d") - activation_name = null_activation_test_data["activation"].name + assert output.startswith("Purged 1 log record") + assert not models.RulebookProcessLog.objects.filter(pk=logs[0].pk).exists() + assert ( + models.RulebookProcessLog.objects.filter( + pk__in=[logs[1].pk, logs[2].pk] + ).count() + == 2 + ) + + +@pytest.mark.django_db +def test_mixed_selectors_form_union_and_validate_all_before_delete( + prepare_log_records, +): + first, second = prepare_log_records + cutoff = (timezone.now() - timedelta(days=5)).strftime("%Y-%m-%d") + original_ids = set( + models.RulebookProcessLog.objects.values_list("id", flat=True) + ) + + with pytest.raises(CommandError, match="Unknown activation names"): + call_command( + "purge_log_records", + "--activation-ids", + str(first.id), + "--activation-names", + second.name, + "missing-name", + "--date", + cutoff, + ) + assert set( + models.RulebookProcessLog.objects.values_list("id", flat=True) + ) == (original_ids) - call_command( + output = call_command( "purge_log_records", + "--activation-ids", + str(first.id), "--activation-names", - activation_name, + second.name, "--date", - cutoff_date, - verbosity=3, + cutoff, + "--audit-trail=false", ) + assert output.startswith("Purged 8 log records") + assert "for selected activations" in output + assert not models.RulebookProcessLog.objects.exists() - captured = capsys.readouterr() - # Check what logs remain - remaining_test_logs = models.RulebookProcessLog.objects.filter( - log__contains="for null test" - ).exclude(log__contains="purged") - - # Verify the fix: NULL activation logs should now be purged - # with name filtering too - null_activation_logs = [ - log - for log in remaining_test_logs - if log.activation_instance.activation_id is None - ] +@pytest.mark.django_db +@pytest.mark.parametrize( + "options, error", + [ + ({"activation-ids": []}, "Activation IDs must"), + ({"activation-ids": [0]}, "Activation IDs must"), + ({"activation-names": [" "]}, "Activation names cannot"), + ], +) +def test_invalid_programmatic_selectors_never_delete( + prepare_log_records, options, error +): + with pytest.raises(CommandError, match=error): + call_command("purge_log_records", date="2024-10-01", **options) + assert models.RulebookProcessLog.objects.count() == 8 + + +@pytest.mark.django_db +@pytest.mark.parametrize( + "args", + [ + ("--audit-trail",), + ("--no-audit-trail",), + ("--audit-trail=maybe",), + ], +) +def test_audit_cli_rejects_unsupported_forms(prepare_log_records, args): + with pytest.raises(CommandError): + call_command("purge_log_records", "--date", "2024-10-01", *args) + assert models.RulebookProcessLog.objects.count() == 8 + + +@pytest.mark.django_db +def test_zero_row_result_creates_no_marker(default_activation_instance): + log = models.RulebookProcessLog.objects.create( + log="newer-than-cutoff", + activation_instance=default_activation_instance, + log_timestamp=int(timezone.now().timestamp()), + ) + output = call_command( + "purge_log_records", "--date", "2024-01-01", "--audit-trail=true" + ) + + assert output.startswith("Purged 0 log records") + assert "globally" in output + assert models.RulebookProcessLog.objects.get(pk=log.pk).log == ( + "newer-than-cutoff" + ) + assert models.RulebookProcessLog.objects.count() == 1 + + +@pytest.mark.django_db +@pytest.mark.parametrize( + "invalid_id", ["", "bad id", "bad|id", "bad\nid", "école", "a" * 129] +) +def test_invalid_internal_job_id_rejected_before_deletion( + default_activation_instance, invalid_id +): + log = models.RulebookProcessLog.objects.create( + log="old", + activation_instance=default_activation_instance, + log_timestamp=1, + ) + with pytest.raises(CommandError, match="must be a UUID or a custom ID"): + call_command( + "purge_log_records", + date="2024-01-01", + **{"internal-job-uuid": invalid_id}, + ) + assert models.RulebookProcessLog.objects.filter(pk=log.pk).exists() + + +@pytest.mark.django_db +def test_cli_valid_internal_uuid_is_in_default_audit_marker( + default_activation_instance, +): + models.RulebookProcessLog.objects.create( + log="old", + activation_instance=default_activation_instance, + log_timestamp=1, + ) + job_id = uuid4() + + output = call_command( + "purge_log_records", + "--date", + "2024-01-01", + "--audit-trail=true", + "--internal-job-uuid", + str(job_id).upper(), + ) + + assert output.startswith("Purged 1 log record") + marker = _audit_markers().get() + assert f"Purge Job ID: {job_id}" in marker.log + assert "Purged 1 log record older than" in marker.log + assert "[Recorded at:" in marker.log + + +@pytest.mark.django_db +def test_cli_custom_internal_job_id_is_in_audit_marker( + default_activation_instance, +): + models.RulebookProcessLog.objects.create( + log="old", + activation_instance=default_activation_instance, + log_timestamp=1, + ) + + output = call_command( + "purge_log_records", + "--date", + "2024-01-01", + "--internal-job-uuid", + "Freds-Troubleshootin-Logs", + ) + + assert output.startswith("Purged 1 log record") + marker = _audit_markers().get() + assert "Purge Job ID: Freds-Troubleshootin-Logs" in marker.log + + +@pytest.mark.django_db +def test_prior_markers_are_deleted_when_older_than_cutoff( + default_activation_instance, +): + process = default_activation_instance + prior_recorded_at = datetime(2024, 2, 1, tzinfo=datetime_timezone.utc) + old_marker = models.RulebookProcessLog.objects.create( + activation_instance=process, + log=( + "Purged 1 log record older than 2024-01-01 00:00:00 UTC. " + "[Recorded at: 2024-02-01 00:00:00 UTC]" + ), + log_timestamp=int(prior_recorded_at.timestamp()), + log_created_at=prior_recorded_at, + ) + ordinary = models.RulebookProcessLog.objects.create( + activation_instance=process, log="ordinary", log_timestamp=1 + ) + cutoff = datetime(2024, 3, 1, tzinfo=datetime_timezone.utc) + + assert delete_log_util.delete_logs_older_than(cutoff) == 2 + first_marker = _audit_markers().get() + assert not models.RulebookProcessLog.objects.filter( + pk=ordinary.pk + ).exists() + assert not models.RulebookProcessLog.objects.filter( + pk=old_marker.pk + ).exists() + assert "Purged 2 log records older than" in first_marker.log + + models.RulebookProcessLog.objects.filter(pk=first_marker.pk).update( + log_timestamp=1 + ) + assert delete_log_util.delete_logs_older_than(cutoff) == 1 + assert not models.RulebookProcessLog.objects.filter( + pk=first_marker.pk + ).exists() + replacement = _audit_markers().get() + assert "Purged 1 log record older than" in replacement.log + assert models.RulebookProcessLog.objects.count() == 1 + + +@pytest.mark.django_db +def test_marker_uses_invocation_count_job_uuid_tracking_and_ceiled_timestamp( + default_activation_instance, +): + process = default_activation_instance + activation = process.activation + activation.log_tracking_id = " tracking-123 " + activation.save(update_fields=["log_tracking_id"]) + models.RulebookProcessLog.objects.create( + activation_instance=process, log="old", log_timestamp=1 + ) + job_id = uuid4() + cutoff = datetime( + 2024, 1, 1, 0, 0, 0, 500000, tzinfo=datetime_timezone.utc + ) + + before_purge = timezone.now() + assert ( + delete_log_util.delete_logs_older_than( + cutoff, internal_job_uuid=str(job_id) + ) + == 1 + ) + after_purge = timezone.now() + marker = _audit_markers().get() + assert ( + max(ceil(before_purge.timestamp()), ceil(cutoff.timestamp())) + <= marker.log_timestamp + <= max(ceil(after_purge.timestamp()), ceil(cutoff.timestamp())) + ) + assert before_purge <= marker.log_created_at <= after_purge + assert marker.log.startswith( + "Purged 1 log record older than " + "2024-01-01 00:00:00.500000 UTC. [Recorded at: " + ) + assert marker.log.endswith( + f" | Purge Job ID: {job_id} | " + "Activation Lifecycle ID: tracking-123]" + ) + + models.RulebookProcessLog.objects.create( + activation_instance=process, log="another old row", log_timestamp=1 + ) + assert delete_log_util.delete_logs_older_than(cutoff) == 1 + markers = list(_audit_markers()) + assert len(markers) == 2 + assert all("Purged 1 log record older than" in row.log for row in markers) + assert ( + "Purge Job ID:" + not in next(row for row in markers if row.pk != marker.pk).log + ) + + +@pytest.mark.django_db +def test_purge_all_logs_replaces_old_rows_and_markers_per_process( + null_activation_test_data, +): + normal = null_activation_test_data["normal_process"] + orphan = null_activation_test_data["null_process"] + prior_recorded_at = datetime(2024, 2, 1, tzinfo=datetime_timezone.utc) + models.RulebookProcessLog.objects.create( + activation_instance=normal, + log=( + "Purged 1 log record older than 2024-01-01 00:00:00 UTC. " + "[Recorded at: 2024-02-01 00:00:00 UTC]" + ), + log_timestamp=int(prior_recorded_at.timestamp()), + log_created_at=prior_recorded_at, + ) + models.RulebookProcessLog.objects.create( + activation_instance=normal, + log="future ordinary log", + log_timestamp=int(timezone.now().timestamp()) + 10000, + ) + models.RulebookProcessLog.objects.create( + activation_instance=orphan, + log=( + "Purged 1 log record older than 2024-01-01 00:00:00 UTC. " + "[Recorded at: 2024-02-01 00:00:00 UTC]" + ), + log_timestamp=int(prior_recorded_at.timestamp()), + log_created_at=prior_recorded_at, + ) + old_ids = set( + models.RulebookProcessLog.objects.values_list("id", flat=True) + ) + job_id = uuid4() + + output = call_command( + "purge_log_records", + "--purge-all-logs", + "--internal-job-uuid", + str(job_id), + ) - # This validates the fix works with activation name filtering - assert len(null_activation_logs) == 0, ( - "All NULL activation logs should be purged " - "when using --activation-names filter" + assert output.startswith("Purged 7 log records globally") + assert not models.RulebookProcessLog.objects.filter( + pk__in=old_ids + ).exists() + markers = models.RulebookProcessLog.objects.all() + assert markers.count() == 2 + normal_marker = markers.get(activation_instance=normal) + orphan_marker = markers.get(activation_instance=orphan) + assert "Purged 4 log records globally. [Recorded at:" in normal_marker.log + assert "Purged 3 log records globally. [Recorded at:" in orphan_marker.log + assert f"Purge Job ID: {job_id}" in normal_marker.log + assert "Activation Lifecycle ID: unavailable (orphaned process)" in ( + orphan_marker.log + ) + assert all( + marker.log_created_at is not None + and f"Recorded at: {delete_log_util._utc_text(marker.log_created_at)}" + in marker.log + for marker in markers + ) + + +@pytest.mark.django_db +def test_purge_all_logs_with_date_deletes_old_logs_globally( + default_activation_instances, +): + old_timestamp = datetime( + 2023, 1, 1, tzinfo=datetime_timezone.utc + ).timestamp() + recent_timestamp = datetime( + 2025, 1, 1, tzinfo=datetime_timezone.utc + ).timestamp() + old_logs = [] + recent_logs = [] + for process in default_activation_instances: + old_logs.append( + models.RulebookProcessLog.objects.create( + activation_instance=process, + log="old", + log_timestamp=old_timestamp, + ) + ) + recent_logs.append( + models.RulebookProcessLog.objects.create( + activation_instance=process, + log="recent", + log_timestamp=recent_timestamp, + ) + ) + + output = call_command( + "purge_log_records", "--purge-all-logs", "--date", "2024-01-01" ) - # Verify that the command output mentions orphaned records - assert "Audit trail will include" in captured.out + assert output.startswith( + "Purged 2 log records globally older than 2024-01-01" + ) + assert not models.RulebookProcessLog.objects.filter( + pk__in=[log.pk for log in old_logs] + ).exists() assert ( - "orphaned" in captured.out - ), "Should mention orphaned records in output" + models.RulebookProcessLog.objects.filter( + pk__in=[log.pk for log in recent_logs] + ).count() + == 2 + ) + assert _audit_markers().count() == 2 + assert all( + "Purged 1 log record older than" in marker.log + for marker in _audit_markers() + ) + + +@pytest.mark.django_db +def test_purge_all_logs_with_retention_duration(default_activation_instances): + now = int(timezone.now().timestamp()) + for process in default_activation_instances: + models.RulebookProcessLog.objects.create( + activation_instance=process, + log="old", + log_timestamp=now - 7200, + ) + models.RulebookProcessLog.objects.create( + activation_instance=process, + log="recent", + log_timestamp=now, + ) + + output = call_command( + "purge_log_records", "--purge-all-logs", "--retention-duration", "PT1H" + ) + + assert output.startswith("Purged 2 log records globally older than") + assert not models.RulebookProcessLog.objects.filter(log="old").exists() + assert models.RulebookProcessLog.objects.filter(log="recent").count() == 2 + assert _audit_markers().count() == 2 + + +@pytest.mark.django_db +@pytest.mark.parametrize( + "conflicting_options", + [ + ("--activation-ids", "1"), + ("--activation-names", "example"), + ("--audit-trail=false",), + ], +) +def test_purge_all_logs_rejects_other_modes_before_deletion( + default_activation_instance, conflicting_options +): + log = models.RulebookProcessLog.objects.create( + activation_instance=default_activation_instance, + log="must remain", + log_timestamp=1, + ) + + with pytest.raises(CommandError, match="purge-all-logs"): + call_command( + "purge_log_records", "--purge-all-logs", *conflicting_options + ) + + assert models.RulebookProcessLog.objects.filter(pk=log.pk).exists() + + +@pytest.mark.django_db +def test_purge_all_logs_rejects_programmatic_non_boolean_flag( + default_activation_instance, +): + log = models.RulebookProcessLog.objects.create( + activation_instance=default_activation_instance, + log="must remain", + log_timestamp=1, + ) + + with pytest.raises(CommandError, match="must be a boolean flag"): + call_command("purge_log_records", **{"purge-all-logs": "false"}) + + assert models.RulebookProcessLog.objects.filter(pk=log.pk).exists() + + +@pytest.mark.django_db +def test_purge_all_logs_with_no_rows_creates_no_marker( + default_activation_instance, +): + output = call_command("purge_log_records", "--purge-all-logs") + + assert output.startswith("Purged 0 log records globally") + assert "No audit markers created." in output + assert not models.RulebookProcessLog.objects.exists() + + +@pytest.mark.django_db +def test_purge_all_logs_deletes_logs_without_process(): + assert not models.RulebookProcess.objects.exists() + with connection.cursor() as cursor: + cursor.execute( + "ALTER TABLE core_rulebook_process_log " + "ALTER COLUMN activation_instance_id DROP NOT NULL" + ) + cursor.execute( + "INSERT INTO core_rulebook_process_log " + "(activation_instance_id, log, log_timestamp) " + "SELECT NULL, 'unattached-' || series.n, 1 " + "FROM generate_series(1, 5) AS series(n) RETURNING id" + ) + inserted_ids = [row[0] for row in cursor.fetchall()] + assert len(inserted_ids) == 5 + + output = call_command("purge_log_records", "--purge-all-logs") + + assert output.startswith("Purged 5 log records globally") + assert "No audit markers created." in output + assert not models.RulebookProcessLog.objects.filter( + id__in=inserted_ids + ).exists() + + +@pytest.mark.django_db +def test_purge_all_logs_with_cutoff_filters_unattached_logs(): + cutoff = datetime(2024, 1, 1, tzinfo=datetime_timezone.utc).timestamp() + with connection.cursor() as cursor: + cursor.execute( + "ALTER TABLE core_rulebook_process_log " + "ALTER COLUMN activation_instance_id DROP NOT NULL" + ) + cursor.execute( + "INSERT INTO core_rulebook_process_log " + "(activation_instance_id, log, log_timestamp) " + "VALUES (NULL, 'old unattached', %s), " + "(NULL, 'recent unattached', %s) RETURNING id", + [cutoff - 1, cutoff + 1], + ) + old_id, recent_id = [row[0] for row in cursor.fetchall()] + + output = call_command( + "purge_log_records", "--purge-all-logs", "--date", "2024-01-01" + ) + + assert output.startswith("Purged 1 log record globally older than") + assert not models.RulebookProcessLog.objects.filter(pk=old_id).exists() + assert models.RulebookProcessLog.objects.filter(pk=recent_id).exists() + assert not _audit_markers().exists() + + +@pytest.mark.django_db(transaction=True) +def test_purge_all_logs_failure_keeps_committed_marker_count( + default_activation_instances, monkeypatch +): + models.RulebookProcessLog.objects.bulk_create( + [ + models.RulebookProcessLog( + activation_instance=process, + log=f"old-{index}", + log_timestamp=1, + ) + for index, process in enumerate(default_activation_instances) + ] + ) + original_delete = QuerySet.delete + calls = 0 + + def fail_second_delete(queryset): + nonlocal calls + calls += 1 + if calls == 2: + raise RuntimeError("injected process failure") + return original_delete(queryset) + + monkeypatch.setattr(QuerySet, "delete", fail_second_delete) + with pytest.raises(CommandError, match="failed after 1 committed rows"): + call_command("purge_log_records", "--purge-all-logs") + + assert ( + models.RulebookProcessLog.objects.filter( + log__startswith="old-" + ).count() + == 1 + ) + marker = _audit_markers().get() + assert "Purged 1 log record globally." in marker.log + + +@pytest.mark.django_db(transaction=True) +def test_marker_failure_rolls_back_process_deletion( + default_activation_instances, monkeypatch +): + """If audit marker creation fails, that process's log deletion + rolls back and the error reports only prior committed rows.""" + models.RulebookProcessLog.objects.bulk_create( + [ + models.RulebookProcessLog( + activation_instance=process, + log=f"old-{index}", + log_timestamp=1, + ) + for index, process in enumerate(default_activation_instances) + ] + ) + original_create = models.RulebookProcessLog.objects.create + calls = {"marker": 0} + + def fail_second_marker(**kwargs): + if kwargs.get("log", "").startswith("Purged "): + calls["marker"] += 1 + if calls["marker"] == 2: + raise RuntimeError("injected marker failure") + return original_create(**kwargs) + + monkeypatch.setattr( + models.RulebookProcessLog.objects, + "create", + fail_second_marker, + ) + with pytest.raises(CommandError, match="failed after 1 committed rows"): + call_command("purge_log_records", "--purge-all-logs") + + assert ( + models.RulebookProcessLog.objects.filter( + log__startswith="old-" + ).count() + == 1 + ) + markers = models.RulebookProcessLog.objects.filter( + log__startswith="Purged " + ) + assert markers.count() == 1 + assert "Purged 1 log record globally." in markers.get().log + + +@pytest.mark.django_db(transaction=True) +def test_purge_all_unattached_failure_raises_command_error_with_count( + default_activation_instance, monkeypatch +): + """LogPurgeError from purge_unattached_logs propagates deleted_count.""" + models.RulebookProcessLog.objects.create( + activation_instance=default_activation_instance, + log="old-log", + log_timestamp=1, + ) + + from aap_eda.core.management.commands import purge_log_records as cmd_mod + + monkeypatch.setattr( + cmd_mod, + "purge_unattached_logs", + mock.Mock(side_effect=RuntimeError("disk full")), + ) + + with pytest.raises(CommandError, match="failed after 1 committed rows"): + call_command("purge_log_records", "--purge-all-logs") diff --git a/tests/integration/tasks/test_management.py b/tests/integration/tasks/test_management.py new file mode 100644 index 000000000..e7a2479e2 --- /dev/null +++ b/tests/integration/tasks/test_management.py @@ -0,0 +1,322 @@ +# Copyright 2026 Red Hat, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 + +from datetime import datetime, timezone +from unittest import mock + +import pytest +from dispatcherd.worker.exceptions import DispatcherCancel +from django.core.management.base import CommandError + +from aap_eda.core import models +from aap_eda.core.utils.delete_log_util import LogPurgeError +from aap_eda.tasks.management import ( + LOG_PURGE_TASK_TIMEOUT_SECONDS, + _aap_eda_manage, + submit_log_purge, +) + + +@pytest.mark.django_db +def test_submit_log_purge_uses_default_queue_uuid_and_purge_timeout(): + job_id = "712beb16-b7dd-40c9-aa29-a7f734376c7a" + with mock.patch( + "aap_eda.tasks.management.uuid4", return_value=job_id + ), mock.patch( + "aap_eda.tasks.management.submit_task", + return_value=({"uuid": job_id}, "default"), + ) as submit_task: + assert ( + submit_log_purge( + retention_duration="P1M", + audit_trail=True, + activation_ids=[9], + ) + == job_id + ) + + (task,) = submit_task.call_args.args + assert task is _aap_eda_manage + assert submit_task.call_args.kwargs == { + "args": ( + job_id, + "purge_log_records", + { + "retention_duration": "P1M", + "audit_trail": True, + "activation-ids": [9], + "internal-job-uuid": job_id, + }, + ), + "queue": "default", + "timeout": LOG_PURGE_TASK_TIMEOUT_SECONDS, + "uuid": job_id, + } + + +@pytest.mark.django_db +def test_submit_log_purge_forwards_scope_when_audit_is_disabled(): + job_id = "712beb16-b7dd-40c9-aa29-a7f734376c7a" + with mock.patch( + "aap_eda.tasks.management.uuid4", return_value=job_id + ), mock.patch( + "aap_eda.tasks.management.submit_task", + return_value=({"uuid": job_id}, "default"), + ) as submit_task: + submit_log_purge( + before_date="2020-01-01", + audit_trail=False, + activation_ids=[9], + ) + + assert submit_task.call_args.kwargs["args"][2] == { + "date": "2020-01-01", + "audit_trail": False, + "activation-ids": [9], + "internal-job-uuid": job_id, + } + + +@pytest.mark.django_db +def test_submitted_purge_all_logs_uses_same_command_and_job_id( + default_activation_instance, new_activation_instance +): + prior_recorded_at = datetime(2024, 2, 1, tzinfo=timezone.utc) + models.RulebookProcessLog.objects.create( + activation_instance=default_activation_instance, + log=( + "Purged 1 log record older than 2024-01-01 00:00:00 UTC. " + "[Recorded at: 2024-02-01 00:00:00 UTC]" + ), + log_timestamp=int(prior_recorded_at.timestamp()), + log_created_at=prior_recorded_at, + ) + models.RulebookProcessLog.objects.create( + activation_instance=default_activation_instance, + log="current log", + log_timestamp=9999999999, + ) + models.RulebookProcessLog.objects.create( + activation_instance=new_activation_instance, + log="another log", + log_timestamp=1, + ) + job_id = "712beb16-b7dd-40c9-aa29-a7f734376c7a" + with mock.patch( + "aap_eda.tasks.management.uuid4", return_value=job_id + ), mock.patch( + "aap_eda.tasks.management.submit_task", + return_value=({"uuid": job_id}, "default"), + ) as submit_task: + assert submit_log_purge(purge_all_logs=True) == job_id + + worker_args = submit_task.call_args.kwargs["args"] + assert worker_args[2] == { + "audit_trail": True, + "purge-all-logs": True, + "internal-job-uuid": job_id, + } + result = _aap_eda_manage(*worker_args) + + assert result["status"] == "completed" + assert result["deleted"] == 3 + markers = models.RulebookProcessLog.objects.all() + assert markers.count() == 2 + assert all(f"Purge Job ID: {job_id}" in marker.log for marker in markers) + assert all(marker.log_created_at is not None for marker in markers) + + +@pytest.mark.django_db +def test_worker_invokes_purge_command_and_returns_deleted_count( + default_activation_instance, + new_activation_instance, +): + selected_log = models.RulebookProcessLog.objects.create( + log="selected worker log", + activation_instance=default_activation_instance, + log_timestamp=1, + ) + retained_log = models.RulebookProcessLog.objects.create( + log="unselected worker log", + activation_instance=new_activation_instance, + log_timestamp=1, + ) + job_id = "712beb16-b7dd-40c9-aa29-a7f734376c7a" + + with mock.patch("aap_eda.tasks.management.logger.info") as log_info: + result = _aap_eda_manage( + job_id, + "purge_log_records", + { + "date": "1970-01-02", + "audit_trail": True, + "activation-ids": [default_activation_instance.activation_id], + "internal-job-uuid": job_id, + }, + ) + + assert result == { + "job_id": job_id, + "command": "purge_log_records", + "status": "completed", + "deleted": 1, + } + assert not models.RulebookProcessLog.objects.filter( + pk=selected_log.pk + ).exists() + assert models.RulebookProcessLog.objects.filter( + pk=retained_log.pk + ).exists() + markers = models.RulebookProcessLog.objects.filter( + activation_instance=default_activation_instance, + log__startswith="Purged ", + ) + assert markers.count() == 1 + assert "Purged 1 log record older than" in markers.get().log + assert f"Purge Job ID: {job_id}" in markers.get().log + assert not models.RulebookProcessLog.objects.filter( + activation_instance=new_activation_instance, + log__startswith="Purged ", + ).exists() + assert any( + call.args[0].startswith("Log purge finished") + and call.args[1:] == (job_id, "completed", 1) + for call in log_info.call_args_list + ) + + +@pytest.mark.django_db +def test_worker_valid_cutoff_without_matching_logs_reports_no_op( + default_activation_instance, +): + models.RulebookProcessLog.objects.create( + log="retained", + activation_instance=default_activation_instance, + log_timestamp=1, + ) + with mock.patch("aap_eda.tasks.management.logger.info") as log_info: + result = _aap_eda_manage( + "no-op-job", "purge_log_records", {"date": "1970-01-01"} + ) + + assert result == { + "job_id": "no-op-job", + "command": "purge_log_records", + "status": "no-op", + "deleted": 0, + } + assert models.RulebookProcessLog.objects.filter(log="retained").exists() + assert models.RulebookProcessLog.objects.count() == 1 + assert any( + call.args[0].startswith("Log purge finished") + and call.args[1:] == ("no-op-job", "no-op", 0) + for call in log_info.call_args_list + ) + + +@pytest.mark.django_db +def test_worker_rejects_missing_cutoff_without_deleting( + default_activation_instance, +): + log = models.RulebookProcessLog.objects.create( + log="retained", + activation_instance=default_activation_instance, + log_timestamp=1, + ) + with pytest.raises(CommandError, match="Exactly one"): + _aap_eda_manage("missing-cutoff", "purge_log_records", {}) + + assert models.RulebookProcessLog.objects.filter(pk=log.pk).exists() + + +@pytest.mark.django_db +def test_worker_lock_contention_is_an_observable_skip(): + class NotAcquired: + def __enter__(self): + return False + + def __exit__(self, exc_type, exc, traceback): + return False + + with mock.patch( + "aap_eda.tasks.management.advisory_lock", return_value=NotAcquired() + ), mock.patch( + "aap_eda.tasks.management.call_command" + ) as call_command, mock.patch( + "aap_eda.tasks.management.logger.warning" + ) as log_warning: + result = _aap_eda_manage("locked-job", "purge_log_records", {}) + + assert result["status"] == "skipped" + assert "advisory lock" in result["reason"] + call_command.assert_not_called() + log_warning.assert_called_once() + assert log_warning.call_args.args[1:] == ( + "locked-job", + "purge_log_records", + "another task holds the advisory lock for purge_log_records", + ) + + +@pytest.mark.django_db +@pytest.mark.parametrize( + "failure", [CommandError("partial failure"), DispatcherCancel()] +) +def test_worker_propagates_command_and_timeout_failures(failure): + with mock.patch( + "aap_eda.tasks.management.call_command", side_effect=failure + ), mock.patch( + "aap_eda.tasks.management.logger.exception" + ) as log_exception, pytest.raises( + type(failure) + ): + _aap_eda_manage("failed-job", "purge_log_records", {}) + + log_exception.assert_called_once() + assert log_exception.call_args.args[0].startswith( + "Management command failed" + ) + assert log_exception.call_args.args[1] == "failed-job" + + +@pytest.mark.django_db +def test_worker_logs_committed_partial_count_on_purge_failure(): + failure = CommandError("later batch failed") + failure.__cause__ = LogPurgeError("later batch failed", 1000) + with mock.patch( + "aap_eda.tasks.management.call_command", side_effect=failure + ), mock.patch( + "aap_eda.tasks.management.logger.exception" + ) as log_exception, pytest.raises( + CommandError + ): + _aap_eda_manage( + "partial-job", "purge_log_records", {"date": "2020-01-01"} + ) + + assert "status=failed deleted_count=%d" in log_exception.call_args.args[0] + assert log_exception.call_args.args[1:] == ( + "partial-job", + "purge_log_records", + 1000, + ) + + +@pytest.mark.django_db +def test_worker_keeps_generic_management_command_scope(): + with mock.patch( + "aap_eda.tasks.management.call_command", return_value="finished" + ) as call_command: + result = _aap_eda_manage("generic-job", "check", {"deploy": True}) + + assert result == { + "job_id": "generic-job", + "command": "check", + "status": "completed", + } + call_command.assert_called_once_with("check", no_color=True, deploy=True) diff --git a/tests/integration/test_advisory_lock.py b/tests/integration/test_advisory_lock.py index aa5be6bf7..cc8cab51f 100644 --- a/tests/integration/test_advisory_lock.py +++ b/tests/integration/test_advisory_lock.py @@ -42,6 +42,12 @@ "fn_call": "_sync_project", "fn_args": [1], }, + { + "module_path": "aap_eda.tasks.management", + "fn_mock": "call_command", + "fn_call": "_aap_eda_manage", + "fn_args": ["parallel-job", "test_management_command", {}], + }, ], ids=[ "gather_analytics", @@ -49,6 +55,7 @@ "monitor_project_tasks", "_import_project", "_sync_project", + "_aap_eda_manage", ], ) @pytest.mark.django_db @@ -84,7 +91,7 @@ def _wrapper_call(shared_list): mocks = [stack.enter_context(patch) for patch in patches] fn_mock = mocks[0] - def record_call(void=None): + def record_call(void=None, **kwargs): time.sleep(1) shared_list.append("called")