Hybrid Cloud Test Gen

Hybrid Cloud Test Generation

This skill generates tests for Sentry's hybrid cloud architecture. It covers RPC services, API gateway proxying, outbox patterns, and endpoint silo decorators.

Critical Constraints

ALWAYS use factory methods (self.create_user(), self.create_organization()) — never Model.objects.create().

NEVER wrap factory method calls in assume_test_silo_mode or assume_test_silo_mode_of. Factories are silo-aware and handle silo mode internally. Only use silo mode context managers for direct ORM queries (Model.objects.get/filter/count/exists/delete).

ALWAYS use pytest-style assertions (assert x == y) — never self.assertEqual().

ALWAYS add tests to existing test files rather than creating new ones, unless no file exists for that module.

For cross-silo ORM access: use assume_test_silo_mode_of(Model) when accessing a single model (auto-detects silo). Use assume_test_silo_mode(SiloMode.X) when the block covers multiple models or non-model operations.

Use TestCase for most tests, including those using outbox_runner(). Only use TransactionTestCase when tests need real committed transactions (threading, concurrency, multi-process scenarios).

NEVER use from __future__ import annotations in test files that deal with RPC models.

Step 1: Identify Test Category

Determine which category of HC test to generate based on the user's request:

If the signal is ambiguous, ask the user to clarify which category.

Step 2: Gather Context

Before generating any test:

1. Read the source module being tested. Determine its silo mode by checking for @cell_silo_endpoint, @control_silo_endpoint, local_mode = SiloMode.X, or @cell_silo_model/@control_silo_model decorators.

2. Find the existing test file using the mirror path convention:

• src/sentry/foo/bar.py → tests/sentry/foo/test_bar.py
• src/sentry/foo/services/bar/service.py → tests/sentry/foo/services/test_bar.py
• src/sentry/foo/services/bar/impl.py → tests/sentry/foo/services/test_bar.py

3. Read the existing test file to understand what's already tested, what base classes are used, and what patterns are established.

4. Read source method signatures to understand parameters, return types, and which RPC models are involved.

Step 3: Generate RPC Service Tests

Load references/rpc-service-tests.md for complete templates and patterns.

RPC service tests must cover:

• Silo compatibility: @all_silo_test ensures the service works across all silo modes
• Serialization round-trip: dispatch_to_local_service verifies args/return survive serialization
• Field accuracy: Field-by-field comparison of RPC model against ORM object
• Error handling: Not-found returns, disabled methods, remote exception wrapping
• Cross-silo effects: outbox_runner() + assume_test_silo_mode for propagation checks

Quick Reference — Decorator & Base Class

Step 4: Generate API Gateway Tests

Load references/api-gateway-tests.md for complete templates and patterns.

API gateway tests verify that requests to control-silo endpoints are correctly proxied to the appropriate cell. They must cover:

• Proxy pass-through: Requests forwarded with correct params, headers, body
• Query parameter forwarding: Multi-value params preserved
• Error proxying: Upstream errors forwarded correctly
• Streaming responses: close_streaming_response() for reading proxied response body

Quick Reference — Decorator & Base Class

Step 5: Generate Outbox Pattern Tests

Load references/outbox-tests.md for complete templates and patterns.

Outbox tests verify that cross-silo messages are created, drained, and produce the expected side effects. They must cover:

• Outbox creation: Verify correct outbox records with outbox_context(flush=False)
• Outbox processing: outbox_runner() drains pending messages
• Cross-silo side effects: assume_test_silo_mode_of(Model) to check replica/mapping state
• Idempotency: Draining the same shard twice produces no duplicates

Quick Reference — Decorator & Base Class

Step 6: Generate Endpoint Silo Tests

Load references/endpoint-silo-tests.md for complete templates and patterns.

Endpoint silo tests verify that API endpoints work correctly under their declared silo mode. They must cover:

• Correct silo decorator: Match endpoint → test decorator
• Cross-silo data setup: Create data using factory methods (no silo wrapper needed)
• Permission checks: Verify 401/403 for unauthorized access
• Response accuracy: Verify response body matches expected data

Quick Reference — Decorator Mapping

Step 7: Validate

Before presenting the generated test, verify against this checklist:

• [ ] Correct silo decorator on test class
• [ ] assume_test_silo_mode_of(Model) for single-model ORM access; assume_test_silo_mode(SiloMode.X) for multi-model/non-model ORM blocks
• [ ] Factory methods (self.create_*) are NEVER wrapped in assume_test_silo_mode
• [ ] Factory methods used — never Model.objects.create()
• [ ] pytest-style assertions only (assert x == y)
• [ ] Correct base class (TestCase for most tests; TransactionTestCase only for threading/concurrency)
• [ ] Imports are correct and minimal
• [ ] Test file at correct mirror path
• [ ] Test methods have descriptive names (test_<action>_<scenario>)
• [ ] Run command: pytest -svv --reuse-db tests/sentry/path/to/test_file.py

Key Imports Quick Reference

# Silo decorators
from sentry.testutils.silo import (
    all_silo_test,
    control_silo_test,
    cell_silo_test,
    no_silo_test,
    assume_test_silo_mode,
    assume_test_silo_mode_of,
    create_test_cells,
)

# Base classes
from sentry.testutils.cases import TestCase, TransactionTestCase, APITestCase

# Cross-silo utilities
from sentry.testutils.outbox import outbox_runner
from sentry.testutils.hybrid_cloud import HybridCloudTestMixin
from sentry.silo.base import SiloMode

# RPC testing
from sentry.hybridcloud.rpc.service import dispatch_to_local_service

# API gateway testing
from sentry.testutils.helpers.apigateway import ApiGatewayTestCase, verify_request_params

# Outbox models
from sentry.hybridcloud.models.outbox import ControlOutbox, CellOutbox, outbox_context
from sentry.hybridcloud.outbox.category import OutboxCategory, OutboxScope

Context Manager Quick Reference

# Use ONLY for direct ORM queries — never for factory calls
assume_test_silo_mode(SiloMode.CONTROL)     # Switch to control silo for ORM access
assume_test_silo_mode(SiloMode.CELL)        # Switch to cell silo for ORM access
assume_test_silo_mode_of(ModelClass)        # Switch to silo matching model's silo mode

outbox_runner()                             # Drain all pending outboxes on exit
outbox_context(flush=False)                 # Create outboxes without flushing
override_cells(cells)                       # Override active cell config
override_settings(SILO_MODE=SiloMode.X)     # Override Django settings
override_options({"key": value})            # Override Sentry options

API Gateway Test Reference

Import Block

from urllib.parse import urlencode

import pytest
import responses
from django.test import override_settings
from django.urls import reverse

from sentry.silo.base import SiloLimit, SiloMode
from sentry.testutils.helpers.apigateway import (
    ApiGatewayTestCase,
    verify_request_params,
    verify_request_body,
    verify_request_headers,
    verify_file_body,
)
from sentry.testutils.helpers.response import close_streaming_response
from sentry.testutils.silo import control_silo_test
from sentry.utils import json

Template: Standard API Gateway Test

@control_silo_test(cells=[ApiGatewayTestCase.CELL], include_monolith_run=True)
class Test{Feature}ApiGateway(ApiGatewayTestCase):

    @responses.activate
    def test_proxy_get_with_params(self):
        """Verify GET request is proxied with query parameters intact."""
        query_params = dict(foo="test", bar=["one", "two"])
        headers = dict(example="this")
        responses.add_callback(
            responses.GET,
            f"{self.CELL.address}/organizations/{self.organization.slug}/{endpoint_path}/",
            verify_request_params(query_params, headers),
        )

        base_url = reverse(
            "{url-name}",
            kwargs={"organization_slug": self.organization.slug},
        )
        encoded_params = urlencode(query_params, doseq=True)
        url = f"{base_url}?{encoded_params}"

        with override_settings(MIDDLEWARE=tuple(self.middleware)):
            resp = self.client.get(url, headers=headers)
        assert resp.status_code == 200, resp.content

    @responses.activate
    def test_proxy_post_with_body(self):
        """Verify POST request is proxied with body intact."""
        request_body = {"key": "value", "nested": {"a": 1}}
        headers = {"content-type": "application/json"}
        responses.add_callback(
            responses.POST,
            f"{self.CELL.address}/organizations/{self.organization.slug}/{endpoint_path}/",
            verify_request_body(request_body, headers),
        )

        url = reverse(
            "{url-name}",
            kwargs={"organization_slug": self.organization.slug},
        )
        with override_settings(MIDDLEWARE=tuple(self.middleware)):
            resp = self.client.post(
                url,
                data=json.dumps(request_body),
                content_type="application/json",
                headers=headers,
            )
        assert resp.status_code == 200, resp.content

    @responses.activate
    def test_proxy_error_forwarded(self):
        """Verify upstream errors are forwarded to the client."""
        responses.add(
            responses.GET,
            f"{self.CELL.address}/organizations/{self.organization.slug}/{endpoint_path}/",
            status=400,
            json={"detail": "Bad request"},
        )

        url = reverse(
            "{url-name}",
            kwargs={"organization_slug": self.organization.slug},
        )
        with override_settings(MIDDLEWARE=tuple(self.middleware)):
            resp = self.client.get(url)
        assert resp.status_code == 400

Template: Reading Proxied Response Content

In CONTROL mode, proxied responses are streamed. Use close_streaming_response() to read the body:

    @responses.activate
    def test_proxy_response_content(self):
        """Verify proxied response content is correct."""
        responses.add_callback(
            responses.GET,
            f"{self.CELL.address}/organizations/{self.organization.slug}/{endpoint_path}/",
            verify_request_params({}, {}),
        )

        url = reverse(
            "{url-name}",
            kwargs={"organization_slug": self.organization.slug},
        )
        with override_settings(MIDDLEWARE=tuple(self.middleware)):
            resp = self.client.get(url)
        assert resp.status_code == 200

        # In CONTROL mode, responses are streamed
        if SiloMode.get_current_mode() == SiloMode.MONOLITH:
            resp_json = json.loads(resp.content)
            assert resp_json["proxy"] is False
        else:
            resp_json = json.loads(close_streaming_response(resp))
            assert resp_json["proxy"] is True

Template: SiloLimit Availability Check

    def test_control_only_endpoint_unavailable_in_cell(self):
        """Verify control-only endpoints raise AvailabilityError outside their silo."""
        with pytest.raises(SiloLimit.AvailabilityError):
            self.client.get("/api/0/{control-only-path}/")

Key Patterns

• `ApiGatewayTestCase` sets up a test cell, mock HTTP callbacks, and the API gateway middleware. It extends APITestCase.
• `@control_silo_test(cells=[...], include_monolith_run=True)` runs the test in both CONTROL and MONOLITH modes.
• Every test method MUST use `@responses.activate` because gateway tests mock HTTP calls to the cell address.
• `verify_request_params(params, headers)` is a callback that asserts query params and headers match.
• `verify_request_body(body, headers)` asserts POST body matches.
• `close_streaming_response(resp)` reads a streaming response to bytes — required for proxied responses in CONTROL mode.
• `override_settings(MIDDLEWARE=tuple(self.middleware))` ensures the API gateway middleware is active.
• `self.CELL` is a pre-configured Cell object with address http://us.internal.sentry.io.
• `self.organization` is pre-created in setUp and bound to self.CELL.

Endpoint Silo Test Reference

Import Block

from sentry.testutils.cases import APITestCase
from sentry.testutils.silo import (
    control_silo_test,
    cell_silo_test,
    no_silo_test,
    assume_test_silo_mode,
    assume_test_silo_mode_of,
    create_test_cells,
)
from sentry.silo.base import SiloMode

Decorator Mapping

Match the endpoint's silo decorator to the test's silo decorator:

Template: Cell Silo Endpoint Test

@cell_silo_test
class Test{Endpoint}(APITestCase):
    endpoint = "sentry-api-0-{endpoint-name}"

    def setUp(self):
        super().setUp()
        # Factory calls: no silo wrapper needed
        self.user = self.create_user()
        self.organization = self.create_organization(owner=self.user)
        self.project = self.create_project(organization=self.organization)
        self.login_as(self.user)

    def test_get_success(self):
        """Verify successful GET returns expected data."""
        response = self.get_success_response(
            self.organization.slug,
            self.project.slug,
        )
        assert response.data["id"] == str(self.project.id)

    def test_get_unauthorized(self):
        """Verify unauthenticated request returns 401."""
        self.login_as(self.create_user())  # different user, no access
        self.get_error_response(
            self.organization.slug,
            self.project.slug,
            status_code=403,
        )

    def test_post_creates_resource(self):
        """Verify POST creates the resource."""
        response = self.get_success_response(
            self.organization.slug,
            method="post",
            name="new-resource",
            status_code=201,
        )
        assert response.data["name"] == "new-resource"

Template: Control Silo Endpoint Test

@control_silo_test
class Test{Endpoint}(APITestCase):
    endpoint = "sentry-api-0-{endpoint-name}"

    def setUp(self):
        super().setUp()
        self.user = self.create_user()
        self.login_as(self.user)

    def test_get_success(self):
        """Verify successful GET for control-silo resource."""
        response = self.get_success_response()
        assert response.data["id"] == str(self.user.id)

Template: Endpoint with Cross-Silo Data Verification

@cell_silo_test
class Test{Endpoint}CrossSilo(APITestCase):
    endpoint = "sentry-api-0-{endpoint-name}"

    def setUp(self):
        super().setUp()
        # Factory calls handle silo mode automatically
        self.user = self.create_user()
        self.organization = self.create_organization(owner=self.user)
        self.login_as(self.user)

    def test_response_includes_cross_silo_data(self):
        """Verify response includes data from the other silo."""
        response = self.get_success_response(self.organization.slug)

        # Verify response against ORM data in the other silo
        with assume_test_silo_mode_of({ControlModel}):
            control_obj = {ControlModel}.objects.get(
                organization_id=self.organization.id,
            )
        assert response.data["{field}"] == str(control_obj.id)

Template: Endpoint with Permission Scopes

@cell_silo_test
class Test{Endpoint}Permissions(APITestCase):
    endpoint = "sentry-api-0-{endpoint-name}"

    def setUp(self):
        super().setUp()
        self.user = self.create_user()
        self.organization = self.create_organization(owner=self.user)
        self.login_as(self.user)

    def test_member_cannot_delete(self):
        """Verify members without admin scope get 403."""
        member_user = self.create_user()
        self.create_member(
            organization=self.organization,
            user=member_user,
            role="member",
        )
        self.login_as(member_user)
        self.get_error_response(
            self.organization.slug,
            method="delete",
            status_code=403,
        )

    def test_admin_can_delete(self):
        """Verify admins with correct scope can delete."""
        self.get_success_response(
            self.organization.slug,
            method="delete",
            status_code=204,
        )

Key Patterns

• `APITestCase` is the standard base class for endpoint tests. It provides get_success_response(), get_error_response(), self.client, and self.login_as().
• `endpoint` class attribute should match the URL name registered in urls.py. Enables get_success_response() / get_error_response() helpers.
• Factory calls (self.create_user(), self.create_organization(), etc.) must NEVER be wrapped in assume_test_silo_mode. Factories are silo-aware.
• `assume_test_silo_mode_of(Model)` is only needed when doing direct ORM queries for verification against the response, not for test setup.
• Permission tests should cover at minimum: unauthenticated (401), unauthorized role (403), and authorized (200/201/204).
• Response data uses string IDs (str(obj.id)) for numeric fields — Sentry's API convention.

Outbox Pattern Test Reference

For outbox system architecture, model mixins, categories, signal receivers, and debugging,

see the hybrid-cloud-outboxes skill. This reference covers test generation patterns only.

Import Block

from unittest.mock import Mock, call, patch

import pytest

from sentry.hybridcloud.models.outbox import (
    ControlOutbox,
    CellOutbox,
    outbox_context,
)
from sentry.hybridcloud.outbox.category import OutboxCategory, OutboxScope
from sentry.models.organization import Organization
from sentry.models.organizationmember import OrganizationMember
from sentry.silo.base import SiloMode
from sentry.testutils.cases import TestCase
from sentry.testutils.outbox import outbox_runner
from sentry.testutils.silo import (
    assume_test_silo_mode,
    assume_test_silo_mode_of,
    control_silo_test,
    cell_silo_test,
)

Template: Outbox Creation Verification

@control_silo_test
class Test{Feature}Outbox(TestCase):

    def test_outbox_created_on_save(self):
        """Verify that saving a model creates the expected outbox record."""
        with outbox_context(flush=False):
            {Model}(id=10).outbox_for_update().save()

        assert {OutboxModel}.objects.count() == 1
        outbox = {OutboxModel}.objects.first()
        assert outbox.shard_scope == OutboxScope.{SCOPE}.value
        assert outbox.shard_identifier == 10
        assert outbox.category == OutboxCategory.{CATEGORY}.value

    def test_multiple_outboxes_created(self):
        """Verify multiple outbox records are created for batch operations."""
        with outbox_context(flush=False):
            {Model}(id=10).outbox_for_update().save()
            {Model}(id=20).outbox_for_update().save()

        assert {OutboxModel}.objects.count() == 2

Template: Outbox Processing and Side Effects

class Test{Feature}OutboxProcessing(TestCase):

    def test_outbox_drains_and_produces_side_effect(self):
        """Verify outbox processing produces the expected cross-silo effect."""
        # Create source objects using factories (no silo wrapper needed)
        org = self.create_organization()
        member = self.create_member(
            organization=org,
            user=self.create_user(),
        )

        # Drain outboxes
        with outbox_runner():
            pass

        # Verify cross-silo effect (silo wrapper needed for ORM query)
        with assume_test_silo_mode_of({ReplicaModel}):
            assert {ReplicaModel}.objects.filter(
                organization_id=org.id,
            ).exists()

    def test_outbox_drain_is_idempotent(self):
        """Verify draining the same shard twice produces no duplicates."""
        org = self.create_organization()

        with outbox_runner():
            pass

        with assume_test_silo_mode_of({ReplicaModel}):
            count_after_first = {ReplicaModel}.objects.count()

        # Drain again — should be a no-op
        with outbox_runner():
            pass

        with assume_test_silo_mode_of({ReplicaModel}):
            assert {ReplicaModel}.objects.count() == count_after_first

Template: Outbox Signal Verification

    @patch("sentry.hybridcloud.models.outbox.process_cell_outbox.send")
    def test_outbox_sends_correct_signal(self, mock_send):
        """Verify the outbox signal fires with correct arguments."""
        org = self.create_organization()

        with outbox_context(flush=False):
            Organization(id=org.id).outbox_for_update().save()

        CellOutbox.objects.filter(
            shard_identifier=org.id,
        ).first().drain_shard()

        mock_send.assert_called_with(
            sender=OutboxCategory.{CATEGORY},
            payload=None,
            object_identifier=org.id,
            shard_identifier=org.id,
            shard_scope=OutboxScope.{SCOPE},
        )

Template: Shard Scheduling Verification

    def test_scheduled_shards(self):
        """Verify correct shards are scheduled for processing."""
        org1 = self.create_organization()
        org2 = self.create_organization()

        with outbox_context(flush=False):
            Organization(id=org1.id).outbox_for_update().save()
            Organization(id=org2.id).outbox_for_update().save()

        shards = {
            (row["shard_scope"], row["shard_identifier"])
            for row in CellOutbox.find_scheduled_shards()
        }
        assert shards == {
            (OutboxScope.ORGANIZATION_SCOPE.value, org1.id),
            (OutboxScope.ORGANIZATION_SCOPE.value, org2.id),
        }

Template: Delete Propagation via Outbox

    def test_delete_propagates_via_outbox(self):
        """Verify deleting an object propagates to the other silo via outbox."""
        # Create objects using factories (no silo wrapper needed)
        org = self.create_organization()
        member = self.create_member(
            organization=org,
            user=self.create_user(),
        )

        # Ensure mapping exists first
        with outbox_runner():
            pass
        with assume_test_silo_mode_of({MappingModel}):
            assert {MappingModel}.objects.filter(
                organizationmember_id=member.id,
            ).exists()

        # Delete and drain
        with outbox_runner():
            member.delete()

        # Verify mapping is gone
        with assume_test_silo_mode_of({MappingModel}):
            assert not {MappingModel}.objects.filter(
                organizationmember_id=member.id,
            ).exists()

Key Patterns

• `outbox_context(flush=False)` creates outbox records without processing them. Use to verify outbox creation.
• `outbox_runner()` processes all pending outboxes synchronously. Works with TestCase — no need for TransactionTestCase.
• `assume_test_silo_mode_of(Model)` is preferred for checking a specific model's state cross-silo. Auto-detects the model's silo.
• `assume_test_silo_mode(SiloMode.X)` for blocks accessing multiple models or non-model resources.
• Factory calls (self.create_organization(), etc.) must NEVER be wrapped in assume_test_silo_mode. Factories handle silo mode internally.
• `@control_silo_test` for tests focused on ControlOutbox records. `@cell_silo_test` for CellOutbox.
• Only use `TransactionTestCase` for threading/concurrency tests (e.g., threading.Barrier), not for standard outbox drain tests.
• Outbox drain fixtures can clear state between tests:

  @pytest.fixture(autouse=True, scope="function")
  def setup_clear_outbox():
      with outbox_runner():
          pass

RPC Service Test Reference

For comprehensive RPC service testing guidance — silo compatibility, serialization round-trips, field accuracy, cross-silo effects, and error handling — see the hybrid-cloud-rpc skill, Step 7 (sections 7.1–7.6).

This file provides supplementary quick-reference patterns specific to test generation.

Supplementary Import Block

These imports cover the full set needed across all RPC test patterns. Pick only what you need:

import pytest
from unittest import mock

from sentry.hybridcloud.rpc.service import (
    dispatch_to_local_service,
    dispatch_remote_call,
    RpcDisabledException,
    RpcRemoteException,
)
from sentry.silo.base import SiloMode
from sentry.testutils.cases import TestCase
from sentry.testutils.helpers import override_options
from sentry.testutils.hybrid_cloud import HybridCloudTestMixin
from sentry.testutils.outbox import outbox_runner
from sentry.testutils.silo import (
    all_silo_test,
    assume_test_silo_mode,
    assume_test_silo_mode_of,
    create_test_cells,
)

Corrections to hybrid-cloud-rpc Step 7

The following patterns in the RPC skill's Step 7 should be applied with these adjustments:

1. `TestCase` is sufficient for `outbox_runner()` — only use TransactionTestCase when tests need real committed transactions (threading, concurrency).

2. Never wrap factory calls in `assume_test_silo_mode` — factories are silo-aware. Only wrap direct ORM queries (Model.objects.get/filter/count/exists/delete).

Template: Composite RPC Test Class

Combines silo compat, field accuracy, serialization, and error handling in one class:

@all_silo_test
class Test{ServiceName}Service(TestCase):
    def setUp(self):
        super().setUp()
        self.user = self.create_user()
        self.organization = self.create_organization(owner=self.user)

    def test_{method_name}_returns_result(self):
        result = {service_instance}.{method_name}(
            organization_id=self.organization.id,
        )
        assert result is not None
        assert result.{field} == expected_value

    def test_{method_name}_not_found_returns_none(self):
        result = {service_instance}.{method_name}(
            organization_id=self.organization.id,
            id=99999,
        )
        assert result is None

    def test_{method_name}_field_accuracy(self):
        orm_obj = {OrmModel}.objects.get(id=thing.id)
        rpc_obj = {service_instance}.{method_name}(
            organization_id=self.organization.id,
            id=orm_obj.id,
        )
        assert rpc_obj.id == orm_obj.id
        assert rpc_obj.name == orm_obj.name
        # ... compare every field

    def test_{method_name}_serialization_round_trip(self):
        serial_arguments = {
            "organization_id": self.organization.id,
        }
        result = dispatch_to_local_service(
            "{service_key}",
            "{method_name}",
            serial_arguments,
        )
        assert result["value"] is not None

Template: Cross-Silo Effects with assume_test_silo_mode_of

Prefer assume_test_silo_mode_of(Model) over assume_test_silo_mode(SiloMode.X) when checking a single model:

@all_silo_test(cells=create_test_cells("us"))
class Test{ServiceName}CrossSilo(TestCase, HybridCloudTestMixin):
    def test_{method_name}_creates_mapping(self):
        with outbox_runner():
            result = {service_instance}.{method_name}(
                organization_id=self.organization.id,
            )

        with assume_test_silo_mode_of({MappingModel}):
            mapping = {MappingModel}.objects.get(
                organization_id=self.organization.id,
            )
        assert result.slug == mapping.slug

    def test_{method_name}_triple_equality(self):
        rpc_result = {service_instance}.{method_name}(
            organization_id=self.organization.id,
        )
        with assume_test_silo_mode_of({OrmModel}):
            orm_obj = {OrmModel}.objects.get(id=rpc_result.id)
        with assume_test_silo_mode_of({MappingModel}):
            mapping = {MappingModel}.objects.get(
                organization_id=self.organization.id,
            )
        assert rpc_result.slug == orm_obj.slug == mapping.slug