tracker/clients/python/tests/test_get_results.py at flux-sync · LearningAzure2020/tracker · 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
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
"""Tests for functions that get scan results for a domain"""
import json

from gql import Client

# Avoid confusion with gql Client
import tracker_client.results as tc
import tracker_client.queries as queries

# pylint: disable=no-member


def test_get_all_results_error(mocker, error_message):
    """Test that get_all_results correctly handles error response"""
    mocker.patch(
        "tracker_client.results.execute_query",
        auto_spec=True,
        return_value=error_message,
    )

    client = Client()
    result = tc.get_all_results(client, "foo.bar")

    assert result == json.dumps(
        error_message,
        indent=4,
    )


def test_get_all_results(mocker, all_results_input, all_results_output):
    """Test that get_all_results produces correct output"""
    mocker.patch(
        "tracker_client.results.execute_query",
        auto_spec=True,
        return_value=all_results_input,
    )

    expected_params = {"domain": "foo.bar"}
    client = Client()
    result = tc.get_all_results(client, "foo.bar")

    tc.execute_query.assert_called_once_with(
        client, queries.ALL_RESULTS, expected_params
    )
    assert result == json.dumps(all_results_output, indent=4)


def test_get_web_results_error(mocker, error_message):
    """Test that get_web_results correctly handles error response"""
    mocker.patch(
        "tracker_client.results.execute_query",
        auto_spec=True,
        return_value=error_message,
    )

    client = Client()
    result = tc.get_web_results(client, "foo.bar")

    assert result == json.dumps(
        error_message,
        indent=4,
    )


def test_get_web_results(mocker, web_results_input, web_results_output):
    """Test that get_web_results produces correct output"""
    mocker.patch(
        "tracker_client.results.execute_query",
        auto_spec=True,
        return_value=web_results_input,
    )

    expected_params = {"domain": "foo.bar"}
    client = Client()
    result = tc.get_web_results(client, "foo.bar")

    tc.execute_query.assert_called_once_with(
        client, queries.WEB_RESULTS, expected_params
    )
    assert result == json.dumps(web_results_output, indent=4)


def test_get_email_results_error(mocker, error_message):
    """Test that get_email_results correctly handles error response"""
    mocker.patch(
        "tracker_client.results.execute_query",
        auto_spec=True,
        return_value=error_message,
    )

    client = Client()
    result = tc.get_email_results(client, "foo.bar")

    assert result == json.dumps(
        error_message,
        indent=4,
    )


def test_get_email_results(mocker, email_results_input, email_results_output):
    """Test that get_email_results produces correct output"""
    mocker.patch(
        "tracker_client.results.execute_query",
        auto_spec=True,
        return_value=email_results_input,
    )

    expected_params = {"domain": "foo.bar"}
    client = Client()
    result = tc.get_email_results(client, "foo.bar")

    tc.execute_query.assert_called_once_with(
        client, queries.EMAIL_RESULTS, expected_params
    )
    assert result == json.dumps(email_results_output, indent=4)


def test_get_domain_status_error(mocker, error_message):
    """Test that get_domain_status correctly handles error response"""
    mocker.patch(
        "tracker_client.results.execute_query",
        auto_spec=True,
        return_value=error_message,
    )

    client = Client()
    result = tc.get_domain_status(client, "foo.bar")

    assert result == json.dumps(
        error_message,
        indent=4,
    )


def test_get_domain_status(mocker, domain_status_input, domain_status_output):
    """Test that get_domain_status produces correct output"""
    mocker.patch(
        "tracker_client.results.execute_query",
        auto_spec=True,
        return_value=domain_status_input,
    )

    expected_params = {"domain": "foo.bar"}
    client = Client()
    result = tc.get_domain_status(client, "foo.bar")

    tc.execute_query.assert_called_once_with(
        client, queries.DOMAIN_STATUS, expected_params
    )
    assert result == json.dumps(domain_status_output, indent=4)