tracker/services/scan-queue/tests/test_scan_queue.py at setup-cluster-network-policies · 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
import pytest
from pretend import stub
from scan_queue import Server, dispatch_dns, dispatch_https, dispatch_ssl

test_queues = {
    "https": stub(enqueue=lambda func, payload, retry, job_timeout, result_ttl: None),
    "ssl": stub(enqueue=lambda func, payload, retry, job_timeout, result_ttl: None),
    "dns": stub(enqueue=lambda func, payload, retry, job_timeout, result_ttl: None),
}


@pytest.fixture
def app():
    client = Server("test", queues=test_queues)
    return client


@pytest.fixture
def client(app):
    with app.test_client() as cli:
        yield cli


def test_enqueue_dns(client):
    test_payload = {
        "scan_id": 1,
        "domain": "cyber.gc.ca",
        "selectors": ["selector1._domainkey", "selector2._domainkey"],
    }

    res = client.post("/dns", json=test_payload)

    assert res.data.decode("utf-8") == "DNS scan request enqueued."


def test_enqueue_https(client):
    test_payload = {
        "scan_id": 1,
        "domain": "cyber.gc.ca",
    }

    res = client.post("/https", json=test_payload)

    assert res.data.decode("utf-8") == "HTTPS scan request enqueued."


def test_enqueue_ssl(client):
    test_payload = {
        "scan_id": 1,
        "domain": "cyber.gc.ca",
    }

    res = client.post("/ssl", json=test_payload)

    assert res.data.decode("utf-8") == "SSL scan request enqueued."