test_rpc.py 27.8 KB
Newer Older
1
2
import re

Kevin Morris's avatar
Kevin Morris committed
3
from http import HTTPStatus
4
from typing import List
Kevin Morris's avatar
Kevin Morris committed
5
6
from unittest import mock

7
import orjson
8
import pytest
9
10

from fastapi.testclient import TestClient
Kevin Morris's avatar
Kevin Morris committed
11
from redis.client import Pipeline
12

13
14
15
import aurweb.models.dependency_type as dt
import aurweb.models.relation_type as rt

16
from aurweb import asgi, config, db, rpc, scripts, time
17
from aurweb.models.account_type import USER_ID
18
from aurweb.models.dependency_type import DEPENDS_ID
19
20
21
22
23
24
25
26
from aurweb.models.license import License
from aurweb.models.package import Package
from aurweb.models.package_base import PackageBase
from aurweb.models.package_dependency import PackageDependency
from aurweb.models.package_keyword import PackageKeyword
from aurweb.models.package_license import PackageLicense
from aurweb.models.package_relation import PackageRelation
from aurweb.models.package_vote import PackageVote
27
from aurweb.models.relation_type import PROVIDES_ID
28
from aurweb.models.user import User
Kevin Morris's avatar
Kevin Morris committed
29
from aurweb.redis import redis_connection
30
31


32
33
34
@pytest.fixture
def client() -> TestClient:
    yield TestClient(app=asgi.app)
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
@pytest.fixture
def user(db_test) -> User:
    with db.begin():
        user = db.create(User, Username="test", Email="test@example.org",
                         RealName="Test User 1", Passwd=str(),
                         AccountTypeID=USER_ID)
    yield user


@pytest.fixture
def user2() -> User:
    with db.begin():
        user = db.create(User, Username="user2", Email="user2@example.org",
                         RealName="Test User 2", Passwd=str(),
                         AccountTypeID=USER_ID)
    yield user


@pytest.fixture
def user3() -> User:
    with db.begin():
        user = db.create(User, Username="user3", Email="user3@example.org",
                         RealName="Test User 3", Passwd=str(),
                         AccountTypeID=USER_ID)
    yield user


@pytest.fixture
def packages(user: User, user2: User, user3: User) -> List[Package]:
    output = []

    # Create package records used in our tests.
    with db.begin():
        pkgbase = db.create(PackageBase, Name="big-chungus",
                            Maintainer=user, Packager=user)
        pkg = db.create(Package, PackageBase=pkgbase, Name=pkgbase.Name,
                        Description="Bunny bunny around bunny",
                        URL="https://example.com/")
        output.append(pkg)

        pkgbase = db.create(PackageBase, Name="chungy-chungus",
                            Maintainer=user, Packager=user)
        pkg = db.create(Package, PackageBase=pkgbase, Name=pkgbase.Name,
                        Description="Wubby wubby on wobba wuubu",
                        URL="https://example.com/")
        output.append(pkg)

        pkgbase = db.create(PackageBase, Name="gluggly-chungus",
                            Maintainer=user, Packager=user)
        pkg = db.create(Package, PackageBase=pkgbase, Name=pkgbase.Name,
                        Description="glurrba glurrba gur globba",
                        URL="https://example.com/")
        output.append(pkg)

        pkgbase = db.create(PackageBase, Name="fugly-chungus",
                            Maintainer=user, Packager=user)
93
94

        desc = "A Package belonging to a PackageBase with another name."
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
        pkg = db.create(Package, PackageBase=pkgbase, Name="other-pkg",
                        Description=desc, URL="https://example.com")
        output.append(pkg)

        pkgbase = db.create(PackageBase, Name="woogly-chungus")
        pkg = db.create(Package, PackageBase=pkgbase, Name=pkgbase.Name,
                        Description="wuggla woblabeloop shemashmoop",
                        URL="https://example.com/")
        output.append(pkg)

    # Setup a few more related records on the first package:
    # a license, some keywords and some votes.
    with db.begin():
        lic = db.create(License, Name="GPL")
        db.create(PackageLicense, Package=output[0], License=lic)

        for keyword in ["big-chungus", "smol-chungus", "sizeable-chungus"]:
            db.create(PackageKeyword,
                      PackageBase=output[0].PackageBase,
                      Keyword=keyword)

116
        now = time.utcnow()
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
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
        for user_ in [user, user2, user3]:
            db.create(PackageVote, User=user_,
                      PackageBase=output[0].PackageBase, VoteTS=now)
    scripts.popupdate.run_single(output[0].PackageBase)

    yield output


@pytest.fixture
def depends(packages: List[Package]) -> List[PackageDependency]:
    output = []

    with db.begin():
        dep = db.create(PackageDependency,
                        Package=packages[0],
                        DepTypeID=dt.DEPENDS_ID,
                        DepName="chungus-depends")
        output.append(dep)

        dep = db.create(PackageDependency,
                        Package=packages[1],
                        DepTypeID=dt.DEPENDS_ID,
                        DepName="chungy-depends")
        output.append(dep)

        dep = db.create(PackageDependency,
                        Package=packages[0],
                        DepTypeID=dt.OPTDEPENDS_ID,
                        DepName="chungus-optdepends",
                        DepCondition="=50")
        output.append(dep)

        dep = db.create(PackageDependency,
                        Package=packages[0],
                        DepTypeID=dt.MAKEDEPENDS_ID,
                        DepName="chungus-makedepends")
        output.append(dep)

        dep = db.create(PackageDependency,
                        Package=packages[0],
                        DepTypeID=dt.CHECKDEPENDS_ID,
                        DepName="chungus-checkdepends")
        output.append(dep)

    yield output


@pytest.fixture
def relations(user: User, packages: List[Package]) -> List[PackageRelation]:
    output = []

    with db.begin():
        rel = db.create(PackageRelation,
                        Package=packages[0],
                        RelTypeID=rt.CONFLICTS_ID,
                        RelName="chungus-conflicts")
        output.append(rel)

        rel = db.create(PackageRelation,
                        Package=packages[1],
                        RelTypeID=rt.CONFLICTS_ID,
                        RelName="chungy-conflicts")
        output.append(rel)

        rel = db.create(PackageRelation,
                        Package=packages[0],
                        RelTypeID=rt.PROVIDES_ID,
                        RelName="chungus-provides",
                        RelCondition="<=200")
        output.append(rel)

        rel = db.create(PackageRelation,
                        Package=packages[0],
                        RelTypeID=rt.REPLACES_ID,
                        RelName="chungus-replaces",
                        RelCondition="<=200")
        output.append(rel)

    # Finally, yield the packages.
    yield output


@pytest.fixture(autouse=True)
def setup(db_test):
    # Create some extra package relationships.
    pass
203

204

Kevin Morris's avatar
Kevin Morris committed
205
206
207
208
209
@pytest.fixture
def pipeline():
    redis = redis_connection()
    pipeline = redis.pipeline()

210
211
    # The 'testclient' host is used when requesting the app
    # via fastapi.testclient.TestClient.
Kevin Morris's avatar
Kevin Morris committed
212
213
    pipeline.delete("ratelimit-ws:testclient")
    pipeline.delete("ratelimit:testclient")
214
    pipeline.execute()
Kevin Morris's avatar
Kevin Morris committed
215
216
217
218

    yield pipeline


219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
def test_rpc_documentation(client: TestClient):
    with client as request:
        resp = request.get("/rpc")
    assert resp.status_code == int(HTTPStatus.OK)
    assert "aurweb RPC Interface" in resp.text


def test_rpc_documentation_missing():
    config_get = config.get

    def mock_get(section: str, key: str) -> str:
        if section == "options" and key == "aurwebdir":
            return "/missing"
        return config_get(section, key)

    with mock.patch("aurweb.config.get", side_effect=mock_get):
        config.rehash()
        expr = r"^doc/rpc\.html could not be read$"
        with pytest.raises(OSError, match=expr):
            rpc.documentation()
    config.rehash()


242
243
244
245
246
def test_rpc_singular_info(client: TestClient,
                           user: User,
                           packages: List[Package],
                           depends: List[PackageDependency],
                           relations: List[PackageRelation]):
247
    # Define expected response.
248
    pkg = packages[0]
249
    expected_data = {
250
251
        "version": 5,
        "results": [{
252
253
254
255
256
257
258
            "Name": pkg.Name,
            "Version": pkg.Version,
            "Description": pkg.Description,
            "URL": pkg.URL,
            "PackageBase": pkg.PackageBase.Name,
            "NumVotes": pkg.PackageBase.NumVotes,
            "Popularity": float(pkg.PackageBase.Popularity),
259
            "OutOfDate": None,
260
261
            "Maintainer": user.Username,
            "URLPath": f"/cgit/aur.git/snapshot/{pkg.Name}.tar.gz",
262
263
264
265
266
267
268
            "Depends": ["chungus-depends"],
            "OptDepends": ["chungus-optdepends=50"],
            "MakeDepends": ["chungus-makedepends"],
            "CheckDepends": ["chungus-checkdepends"],
            "Conflicts": ["chungus-conflicts"],
            "Provides": ["chungus-provides<=200"],
            "Replaces": ["chungus-replaces<=200"],
269
            "License": [pkg.package_licenses.first().License.Name],
270
271
272
273
274
275
276
277
278
            "Keywords": [
                "big-chungus",
                "sizeable-chungus",
                "smol-chungus"
            ]
        }],
        "resultcount": 1,
        "type": "multiinfo"
    }
279
280

    # Make dummy request.
281
    with client as request:
282
283
284
285
286
        resp = request.get("/rpc", params={
            "v": 5,
            "type": "info",
            "arg": ["chungy-chungus", "big-chungus"],
        })
287
288

    # Load  request response into Python dictionary.
289
    response_data = orjson.loads(resp.text)
290
291
292
293
294

    # Remove the FirstSubmitted LastModified, ID and PackageBaseID keys from
    # reponse, as the key's values aren't guaranteed to match between the two
    # (the keys are already removed from 'expected_data').
    for i in ["FirstSubmitted", "LastModified", "ID", "PackageBaseID"]:
295
        response_data["results"][0].pop(i)
296
297

    # Validate that the new dictionaries are the same.
298
    assert response_data == expected_data
299
300


301
def test_rpc_nonexistent_package(client: TestClient):
302
    # Make dummy request.
303
304
    with client as request:
        response = request.get("/rpc/?v=5&type=info&arg=nonexistent-package")
305
306
307
308
309
310
311
312

    # Load request response into Python dictionary.
    response_data = orjson.loads(response.content.decode())

    # Validate data.
    assert response_data["resultcount"] == 0


313
def test_rpc_multiinfo(client: TestClient, packages: List[Package]):
314
315
    # Make dummy request.
    request_packages = ["big-chungus", "chungy-chungus"]
316
    with client as request:
317
318
319
        response = request.get("/rpc", params={
            "v": 5, "type": "info", "arg[]": request_packages
        })
320
321
322
323
324
325
326
327
328
329
330

    # Load request response into Python dictionary.
    response_data = orjson.loads(response.content.decode())

    # Validate data.
    for i in response_data["results"]:
        request_packages.remove(i["Name"])

    assert request_packages == []


331
def test_rpc_mixedargs(client: TestClient, packages: List[Package]):
332
333
334
335
    # Make dummy request.
    response1_packages = ["gluggly-chungus"]
    response2_packages = ["gluggly-chungus", "chungy-chungus"]

336
    with client as request:
337
        # Supply all of the args in the url to enforce ordering.
338
339
340
341
342
343
        response1 = request.get(
            "/rpc?v=5&arg[]=big-chungus&arg=gluggly-chungus&type=info")
    assert response1.status_code == int(HTTPStatus.OK)

    with client as request:
        response2 = request.get(
344
345
            "/rpc?v=5&arg=big-chungus&arg[]=gluggly-chungus"
            "&type=info&arg[]=chungy-chungus")
346
    assert response1.status_code == int(HTTPStatus.OK)
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362

    # Load request response into Python dictionary.
    response1_data = orjson.loads(response1.content.decode())
    response2_data = orjson.loads(response2.content.decode())

    # Validate data.
    for i in response1_data["results"]:
        response1_packages.remove(i["Name"])

    for i in response2_data["results"]:
        response2_packages.remove(i["Name"])

    for i in [response1_packages, response2_packages]:
        assert i == []


363
364
365
366
367
368
369
370
371
def test_rpc_no_dependencies_omits_key(client: TestClient, user: User,
                                       packages: List[Package],
                                       depends: List[PackageDependency],
                                       relations: List[PackageRelation]):
    """
    This makes sure things like 'MakeDepends' get removed from JSON strings
    when they don't have set values.
    """
    pkg = packages[1]
372
    expected_response = {
373
374
        'version': 5,
        'results': [{
375
376
377
378
379
380
381
            'Name': pkg.Name,
            'Version': pkg.Version,
            'Description': pkg.Description,
            'URL': pkg.URL,
            'PackageBase': pkg.PackageBase.Name,
            'NumVotes': pkg.PackageBase.NumVotes,
            'Popularity': int(pkg.PackageBase.Popularity),
382
            'OutOfDate': None,
383
            'Maintainer': user.Username,
384
385
386
387
388
389
390
391
392
            'URLPath': '/cgit/aur.git/snapshot/chungy-chungus.tar.gz',
            'Depends': ['chungy-depends'],
            'Conflicts': ['chungy-conflicts'],
            'License': [],
            'Keywords': []
        }],
        'resultcount': 1,
        'type': 'multiinfo'
    }
393
394

    # Make dummy request.
395
    with client as request:
396
397
398
        response = request.get("/rpc", params={
            "v": 5, "type": "info", "arg": "chungy-chungus"
        })
399
400
401
402
403
404
405
406
407
    response_data = orjson.loads(response.content.decode())

    # Remove inconsistent keys.
    for i in ["ID", "PackageBaseID", "FirstSubmitted", "LastModified"]:
        response_data["results"][0].pop(i)

    assert response_data == expected_response


408
def test_rpc_bad_type(client: TestClient):
409
410
    # Define expected response.
    expected_data = {
411
412
413
414
415
416
        'version': 5,
        'results': [],
        'resultcount': 0,
        'type': 'error',
        'error': 'Incorrect request type specified.'
    }
417
418

    # Make dummy request.
419
    with client as request:
420
421
422
        response = request.get("/rpc", params={
            "v": 5, "type": "invalid-type", "arg": "big-chungus"
        })
423
424
425
426
427
428
429
430

    # Load  request response into Python dictionary.
    response_data = orjson.loads(response.content.decode())

    # Validate data.
    assert expected_data == response_data


431
def test_rpc_bad_version(client: TestClient):
432
    # Define expected response.
433
434
435
436
437
438
439
    expected_data = {
        'version': 0,
        'resultcount': 0,
        'results': [],
        'type': 'error',
        'error': 'Invalid version specified.'
    }
440
441

    # Make dummy request.
442
    with client as request:
443
444
445
        response = request.get("/rpc", params={
            "v": 0, "type": "info", "arg": "big-chungus"
        })
446
447
448
449
450
451
452
453

    # Load  request response into Python dictionary.
    response_data = orjson.loads(response.content.decode())

    # Validate data.
    assert expected_data == response_data


454
def test_rpc_no_version(client: TestClient):
455
    # Define expected response.
456
457
458
459
460
461
462
    expected_data = {
        'version': None,
        'resultcount': 0,
        'results': [],
        'type': 'error',
        'error': 'Please specify an API version.'
    }
463
464

    # Make dummy request.
465
    with client as request:
466
467
468
469
        response = request.get("/rpc", params={
            "type": "info",
            "arg": "big-chungus"
        })
470
471
472
473
474
475
476
477

    # Load  request response into Python dictionary.
    response_data = orjson.loads(response.content.decode())

    # Validate data.
    assert expected_data == response_data


478
def test_rpc_no_type(client: TestClient):
479
480
    # Define expected response.
    expected_data = {
481
482
483
484
485
486
        'version': 5,
        'results': [],
        'resultcount': 0,
        'type': 'error',
        'error': 'No request type/data specified.'
    }
487
488

    # Make dummy request.
489
    with client as request:
490
        response = request.get("/rpc", params={"v": 5, "arg": "big-chungus"})
491
492
493
494
495
496
497
498

    # Load  request response into Python dictionary.
    response_data = orjson.loads(response.content.decode())

    # Validate data.
    assert expected_data == response_data


499
def test_rpc_no_args(client: TestClient):
500
501
    # Define expected response.
    expected_data = {
502
503
504
505
506
507
        'version': 5,
        'results': [],
        'resultcount': 0,
        'type': 'error',
        'error': 'No request type/data specified.'
    }
508
509

    # Make dummy request.
510
    with client as request:
511
        response = request.get("/rpc", params={"v": 5, "type": "info"})
512
513
514
515
516
517
518
519

    # Load  request response into Python dictionary.
    response_data = orjson.loads(response.content.decode())

    # Validate data.
    assert expected_data == response_data


520
def test_rpc_no_maintainer(client: TestClient, packages: List[Package]):
521
    # Make dummy request.
522
    with client as request:
523
524
525
        response = request.get("/rpc", params={
            "v": 5, "type": "info", "arg": "woogly-chungus"
        })
526
527
528
529
530
531

    # Load  request response into Python dictionary.
    response_data = orjson.loads(response.content.decode())

    # Validate data.
    assert response_data["results"][0]["Maintainer"] is None
532
533


534
535
def test_rpc_suggest_pkgbase(client: TestClient, packages: List[Package]):
    params = {"v": 5, "type": "suggest-pkgbase", "arg": "big"}
536
    with client as request:
537
        response = request.get("/rpc", params=params)
538
539
540
    data = response.json()
    assert data == ["big-chungus"]

541
    params["arg"] = "chungy"
542
    with client as request:
543
        response = request.get("/rpc", params=params)
544
545
    data = response.json()
    assert data == ["chungy-chungus"]
546

547
    # Test no arg supplied.
548
    del params["arg"]
549
    with client as request:
550
        response = request.get("/rpc", params=params)
551
552
553
    data = response.json()
    assert data == []

554
555
556
557
558
559
560
561
    # Test that suggestions are only given based on the beginning
    # of the keyword string.
    params["arg"] = "ther-pkg"
    with client as request:
        response = request.get("/rpc", params=params)
    data = response.json()
    assert data == []

562

563
564
def test_rpc_suggest(client: TestClient, packages: List[Package]):
    params = {"v": 5, "type": "suggest", "arg": "other"}
565
    with client as request:
566
        response = request.get("/rpc", params=params)
567
568
569
570
    data = response.json()
    assert data == ["other-pkg"]

    # Test non-existent Package.
571
    params["arg"] = "nonexistent"
572
    with client as request:
573
        response = request.get("/rpc", params=params)
574
575
576
    data = response.json()
    assert data == []

577
    # Test no arg supplied.
578
    del params["arg"]
579
    with client as request:
580
        response = request.get("/rpc", params=params)
581
582
583
    data = response.json()
    assert data == []

584
585
586
587
588
589
590
591
    # Test that suggestions are only given based on the beginning
    # of the keyword string.
    params["arg"] = "ther-pkg"
    with client as request:
        response = request.get("/rpc", params=params)
    data = response.json()
    assert data == []

592

Kevin Morris's avatar
Kevin Morris committed
593
594
595
596
597
598
599
600
601
def mock_config_getint(section: str, key: str):
    if key == "request_limit":
        return 4
    elif key == "window_length":
        return 100
    return config.getint(section, key)


@mock.patch("aurweb.config.getint", side_effect=mock_config_getint)
602
def test_rpc_ratelimit(getint: mock.MagicMock, client: TestClient,
603
604
605
                       pipeline: Pipeline, packages: List[Package]):
    params = {"v": 5, "type": "suggest-pkgbase", "arg": "big"}

Kevin Morris's avatar
Kevin Morris committed
606
607
    for i in range(4):
        # The first 4 requests should be good.
608
        with client as request:
609
            response = request.get("/rpc", params=params)
Kevin Morris's avatar
Kevin Morris committed
610
611
612
        assert response.status_code == int(HTTPStatus.OK)

    # The fifth request should be banned.
613
    with client as request:
614
        response = request.get("/rpc", params=params)
Kevin Morris's avatar
Kevin Morris committed
615
616
617
618
619
620
621
622
623
    assert response.status_code == int(HTTPStatus.TOO_MANY_REQUESTS)

    # Delete the cached records.
    pipeline.delete("ratelimit-ws:testclient")
    pipeline.delete("ratelimit:testclient")
    one, two = pipeline.execute()
    assert one and two

    # The new first request should be good.
624
    with client as request:
625
        response = request.get("/rpc", params=params)
Kevin Morris's avatar
Kevin Morris committed
626
    assert response.status_code == int(HTTPStatus.OK)
627
628


629
630
def test_rpc_etag(client: TestClient, packages: List[Package]):
    params = {"v": 5, "type": "suggest-pkgbase", "arg": "big"}
631
632

    with client as request:
633
634
635
636
        response1 = request.get("/rpc", params=params)
    with client as request:
        response2 = request.get("/rpc", params=params)

637
638
639
    assert response1.headers.get("ETag") is not None
    assert response1.headers.get("ETag") != str()
    assert response1.headers.get("ETag") == response2.headers.get("ETag")
640
641


642
def test_rpc_search_arg_too_small(client: TestClient):
643
    params = {"v": 5, "type": "search", "arg": "b"}
644
    with client as request:
645
        response = request.get("/rpc", params=params)
646
647
648
649
    assert response.status_code == int(HTTPStatus.OK)
    assert response.json().get("error") == "Query arg too small."


650
651
def test_rpc_search(client: TestClient, packages: List[Package]):
    params = {"v": 5, "type": "search", "arg": "big"}
652
    with client as request:
653
        response = request.get("/rpc", params=params)
654
655
656
657
658
659
    assert response.status_code == int(HTTPStatus.OK)

    data = response.json()
    assert data.get("resultcount") == 1

    result = data.get("results")[0]
660
    assert result.get("Name") == packages[0].Name
661

662
663
664
    # Test the If-None-Match headers.
    etag = response.headers.get("ETag").strip('"')
    headers = {"If-None-Match": etag}
665
    response = request.get("/rpc", params=params, headers=headers)
666
667
668
    assert response.status_code == int(HTTPStatus.NOT_MODIFIED)
    assert response.content == b''

669
    # No args on non-m by types return an error.
670
671
672
    del params["arg"]
    with client as request:
        response = request.get("/rpc", params=params)
673
674
675
    assert response.json().get("error") == "No request type/data specified."


676
677
def test_rpc_msearch(client: TestClient, user: User, packages: List[Package]):
    params = {"v": 5, "type": "msearch", "arg": user.Username}
678
    with client as request:
679
        response = request.get("/rpc", params=params)
680
681
682
683
684
    data = response.json()

    # user1 maintains 4 packages; assert that we got them all.
    assert data.get("resultcount") == 4
    names = list(sorted(r.get("Name") for r in data.get("results")))
685
    expected_results = [
686
687
688
689
        "big-chungus",
        "chungy-chungus",
        "gluggly-chungus",
        "other-pkg"
690
    ]
691
692
693
    assert names == expected_results

    # Search for a non-existent maintainer, giving us zero packages.
694
695
    params["arg"] = "blah-blah"
    response = request.get("/rpc", params=params)
696
697
698
    data = response.json()
    assert data.get("resultcount") == 0

699
700
701
    with db.begin():
        packages[0].PackageBase.Maintainer = None

702
703
    # A missing arg still succeeds, but it returns all orphans.
    # Just verify that we receive no error and the orphaned result.
704
705
    params.pop("arg")
    response = request.get("/rpc", params=params)
706
707
708
    data = response.json()
    assert data.get("resultcount") == 1
    result = data.get("results")[0]
709
    assert result.get("Name") == "big-chungus"
710
711


712
713
714
715
716
def test_rpc_search_depends(client: TestClient, packages: List[Package],
                            depends: List[PackageDependency]):
    params = {
        "v": 5, "type": "search", "by": "depends", "arg": "chungus-depends"
    }
717
    with client as request:
718
        response = request.get("/rpc", params=params)
719
720
721
    data = response.json()
    assert data.get("resultcount") == 1
    result = data.get("results")[0]
722
    assert result.get("Name") == packages[0].Name
723
724


725
726
727
728
729
730
731
732
def test_rpc_search_makedepends(client: TestClient, packages: List[Package],
                                depends: List[PackageDependency]):
    params = {
        "v": 5,
        "type": "search",
        "by": "makedepends",
        "arg": "chungus-makedepends"
    }
733
    with client as request:
734
        response = request.get("/rpc", params=params)
735
736
737
    data = response.json()
    assert data.get("resultcount") == 1
    result = data.get("results")[0]
738
    assert result.get("Name") == packages[0].Name
739
740


741
742
743
744
745
746
747
748
def test_rpc_search_optdepends(client: TestClient, packages: List[Package],
                               depends: List[PackageDependency]):
    params = {
        "v": 5,
        "type": "search",
        "by": "optdepends",
        "arg": "chungus-optdepends"
    }
749
    with client as request:
750
        response = request.get("/rpc", params=params)
751
752
753
    data = response.json()
    assert data.get("resultcount") == 1
    result = data.get("results")[0]
754
    assert result.get("Name") == packages[0].Name
755
756


757
758
759
760
761
762
763
764
def test_rpc_search_checkdepends(client: TestClient, packages: List[Package],
                                 depends: List[PackageDependency]):
    params = {
        "v": 5,
        "type": "search",
        "by": "checkdepends",
        "arg": "chungus-checkdepends"
    }
765
    with client as request:
766
        response = request.get("/rpc", params=params)
767
768
769
    data = response.json()
    assert data.get("resultcount") == 1
    result = data.get("results")[0]
770
    assert result.get("Name") == packages[0].Name
771
772


773
def test_rpc_incorrect_by(client: TestClient):
774
    params = {"v": 5, "type": "search", "by": "fake", "arg": "big"}
775
    with client as request:
776
        response = request.get("/rpc", params=params)
777
    assert response.json().get("error") == "Incorrect by field specified."
778
779


780
def test_rpc_jsonp_callback(client: TestClient):
781
782
783
784
785
    """ Test the callback parameter.

    For end-to-end verification, the `examples/jsonp.html` file can be
    used to submit jsonp callback requests to the RPC.
    """
786
787
788
789
790
791
    params = {
        "v": 5,
        "type": "search",
        "arg": "big",
        "callback": "jsonCallback"
    }
792
    with client as request:
793
        response = request.get("/rpc", params=params)
794
795
    assert response.headers.get("content-type") == "text/javascript"
    assert re.search(r'^/\*\*/jsonCallback\(.*\)$', response.text) is not None
796
797

    # Test an invalid callback name; we get an application/json error.
798
    params["callback"] = "jsonCallback!"
799
    with client as request:
800
        response = request.get("/rpc", params=params)
801
802
    assert response.headers.get("content-type") == "application/json"
    assert response.json().get("error") == "Invalid callback name."
Kevin Morris's avatar
Kevin Morris committed
803
804
805
806
807
808
809
810
811
812
813
814
815


def test_rpc_post(client: TestClient, packages: List[Package]):
    data = {
        "v": 5,
        "type": "info",
        "arg": "big-chungus",
        "arg[]": ["chungy-chungus"]
    }
    with client as request:
        resp = request.post("/rpc", data=data)
    assert resp.status_code == int(HTTPStatus.OK)
    assert resp.json().get("resultcount") == 2
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834


def test_rpc_too_many_search_results(client: TestClient,
                                     packages: List[Package]):
    config_getint = config.getint

    def mock_config(section: str, key: str):
        if key == "max_rpc_results":
            return 1
        return config_getint(section, key)

    params = {"v": 5, "type": "search", "arg": "chungus"}
    with mock.patch("aurweb.config.getint", side_effect=mock_config):
        with client as request:
            resp = request.get("/rpc", params=params)
    assert resp.json().get("error") == "Too many package results."


def test_rpc_too_many_info_results(client: TestClient, packages: List[Package]):
835
836
837
838
839
840
841
842
843
844
    # Make many of these packages depend and rely on each other.
    # This way, we can test to see that the exceeded limit stays true
    # regardless of the number of related records.
    with db.begin():
        for i in range(len(packages) - 1):
            db.create(PackageDependency, DepTypeID=DEPENDS_ID,
                      Package=packages[i], DepName=packages[i + 1].Name)
            db.create(PackageRelation, RelTypeID=PROVIDES_ID,
                      Package=packages[i], RelName=packages[i + 1].Name)

845
846
847
848
849
850
851
852
853
854
855
856
    config_getint = config.getint

    def mock_config(section: str, key: str):
        if key == "max_rpc_results":
            return 1
        return config_getint(section, key)

    params = {"v": 5, "type": "info", "arg[]": [p.Name for p in packages]}
    with mock.patch("aurweb.config.getint", side_effect=mock_config):
        with client as request:
            resp = request.get("/rpc", params=params)
    assert resp.json().get("error") == "Too many package results."