codecov-api/codecov_auth/tests/test_signals.py at 005f2f82f36bba4e6aeb7268bd2a2713e1efcf2f · codecov/codecov-api · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
from unittest import mock
from unittest.mock import call

import pytest
from django.test import TestCase, override_settings
from shared.django_apps.codecov_auth.models import Service
from shared.django_apps.codecov_auth.tests.factories import (
    OrganizationLevelTokenFactory,
    OwnerFactory,
)

from utils.shelter import ShelterPubsub


@pytest.mark.django_db
def test_shelter_org_token_sync(mocker):
    publish = mocker.patch("utils.shelter.ShelterPubsub.publish")

    # this triggers the publish via Django signals
    OrganizationLevelTokenFactory(id=91728376, owner=OwnerFactory(ownerid=111))

    publish.assert_has_calls(
        [
            call({"type": "owner", "sync": "one", "id": 111}),
            call({"type": "org_token", "sync": "one", "id": 91728376}),
        ]
    )


@mock.patch("utils.shelter.ShelterPubsub.publish")
class TestCodecovAuthSignals(TestCase):
    def test_sync_on_create(self, mock_publish):
        OwnerFactory(ownerid=12345)
        mock_publish.assert_called_once_with(
            {"type": "owner", "sync": "one", "id": 12345}
        )

    def test_sync_on_update_upload_token_required_for_public_repos(self, mock_publish):
        owner = OwnerFactory(ownerid=12345, upload_token_required_for_public_repos=True)
        owner.upload_token_required_for_public_repos = False
        owner.save()
        mock_publish.assert_has_calls(
            [
                call({"type": "owner", "sync": "one", "id": 12345}),
                call({"type": "owner", "sync": "one", "id": 12345}),
            ]
        )

    def test_sync_on_update_username(self, mock_publish):
        owner = OwnerFactory(ownerid=12345, username="hello")
        owner.username = "world"
        owner.save()
        mock_publish.assert_has_calls(
            [
                call({"type": "owner", "sync": "one", "id": 12345}),
                call({"type": "owner", "sync": "one", "id": 12345}),
            ]
        )

    def test_sync_on_update_service(self, mock_publish):
        owner = OwnerFactory(ownerid=12345, service=Service.GITHUB.value)
        owner.service = Service.BITBUCKET.value
        owner.save()
        mock_publish.assert_has_calls(
            [
                call({"type": "owner", "sync": "one", "id": 12345}),
                call({"type": "owner", "sync": "one", "id": 12345}),
            ]
        )

    def test_no_sync_on_update_other_fields(self, mock_publish):
        owner = OwnerFactory(ownerid=12345, name="hello")
        owner.name = "world"
        owner.save()
        mock_publish.assert_called_once_with(
            {"type": "owner", "sync": "one", "id": 12345}
        )


@mock.patch("logging.Logger.warning")
@mock.patch("google.cloud.pubsub_v1.PublisherClient.publish")
@mock.patch("utils.shelter.ShelterPubsub.get_instance")
@override_settings(SHELTER_ENABLED=True)
@pytest.mark.django_db
def test_sync_error(mock_instance, mock_publish, mock_log):
    mock_instance.return_value = ShelterPubsub()
    mock_publish.side_effect = Exception("publish error")

    OwnerFactory(ownerid=12345)

    # publish is still called, raises an Exception
    mock_publish.assert_called_once_with(
        "projects/test-project-id/topics/test-topic-id",
        b'{"type": "owner", "sync": "one", "id": 12345}',
    )

    mock_log.assert_called_once_with(
        "Failed to publish a message",
        extra=dict(
            data_to_publish={"type": "owner", "sync": "one", "id": 12345},
            error=mock_publish.side_effect,
        ),
    )