tests/test_play.py (29383 bytes)
1 """Google Play purchases of cloud caption hours: the Play build of Subrep. 2 3 No test reaches Google. FakeGooglePlay replaces play.get, play.consume and 4 play.voided, and the tests of the HTTP layer give play._session a fake session. 5 """ 6 7 from __future__ import annotations 8 9 import errno 10 import itertools 11 import json 12 import logging 13 import os 14 import pathlib 15 import secrets 16 from urllib.parse import unquote 17 18 import pytest 19 import requests 20 from google.auth import exceptions as google_errors 21 from hypothesis import given 22 from hypothesis import settings as hsettings 23 from hypothesis import strategies as st 24 25 from backend import accounts, captions, main, play 26 from backend.db import Account, SessionLocal 27 from backend.settings import settings 28 29 from .conftest import account_id 30 31 HOUR = 3600 32 UNKNOWN = "Google Play does not know this purchase." 33 _orders = itertools.count(1) 34 35 36 class FakeGooglePlay: 37 """Google Play as this server sees it: the purchases of the app, by token.""" 38 39 def __init__(self): 40 self.purchases: dict[str, dict] = {} 41 # The consumes that came after the server committed the ledger row. 42 self.consumed: list[str] = [] 43 # The consumes that came before that commit. The fixture `google` 44 # fails each test that makes one: if the commit fails after the 45 # consume, the buyer paid, and Google closed the purchase, but the 46 # account has no hours. 47 self.early: list[str] = [] 48 self.gets: list[tuple[str, str]] = [] 49 self.down = False 50 self.fail_consume = 0 51 # The tokens of the purchases that Google refunded or charged back. 52 self.voided_tokens: list[str] = [] 53 54 def buy(self, token: str, account: str | None, product: str = "captions20", 55 state: int = 0, quantity: int | None = None, test: bool = False) -> dict: 56 """Add a purchase, as purchases.products.get returns it. 57 58 `account` None is a purchase without an account id, for example with 59 a promo code in the Play Store app. 60 """ 61 purchase = {"kind": "androidpublisher#productPurchase", "productId": product, 62 "purchaseState": state, "consumptionState": 0, 63 "orderId": f"GPA.3300-{next(_orders):04d}"} 64 if account is not None: 65 purchase["obfuscatedExternalAccountId"] = account 66 if quantity is not None: 67 purchase["quantity"] = quantity 68 if test: 69 purchase["purchaseType"] = 0 70 self.purchases[token] = purchase 71 return purchase 72 73 def get(self, product_id: str, token: str) -> dict: 74 self.gets.append((product_id, token)) 75 if self.down: 76 raise play.PlayError(502, play.NO_ANSWER) 77 if token not in self.purchases: 78 raise play.PlayError(400, UNKNOWN) 79 return dict(self.purchases[token]) 80 81 def consume(self, product_id: str, token: str) -> None: 82 if self.fail_consume: 83 self.fail_consume -= 1 84 raise play.PlayError(502, play.NO_ANSWER) 85 (self.consumed if ledger(token) else self.early).append(token) 86 self.purchases[token]["consumptionState"] = 1 87 88 def voided(self) -> list[str]: 89 return list(self.voided_tokens) 90 91 92 @pytest.fixture(autouse=True) 93 def _no_google(monkeypatch): 94 """No test may reach Google. A test of the HTTP layer gives its own fake session.""" 95 def refuse(): 96 raise AssertionError("a test tried to reach Google Play") 97 98 monkeypatch.setattr(play, "_session", refuse) 99 100 101 @pytest.fixture 102 def google(monkeypatch, tmp_path): 103 """A server with a Play key file, and Google Play faked.""" 104 fake = FakeGooglePlay() 105 monkeypatch.setattr(play, "get", fake.get) 106 monkeypatch.setattr(play, "consume", fake.consume) 107 monkeypatch.setattr(play, "voided", fake.voided) 108 key = tmp_path / "play-key.json" 109 key.write_text("{}") # play.get is fake, so nothing reads the key. 110 monkeypatch.setattr(settings, "play_service_account_file", str(key)) 111 yield fake 112 assert fake.early == [], "the server consumed a purchase before it committed its hours" 113 114 115 def new_token() -> str: 116 """A purchase token that no other test uses: the test database lives for the whole run.""" 117 return f"tok.{secrets.token_hex(8)}-_" 118 119 120 def post_play(client, token: str, product: str = "captions20"): 121 return client.post("/api/captions/play-purchase", 122 json={"product_id": product, "purchase_token": token}) 123 124 125 def left(client) -> int: 126 return client.get("/api/captions").json()["seconds_left"] 127 128 129 def ledger(token: str) -> list[play.PlayPurchase]: 130 with SessionLocal() as s: 131 return s.query(play.PlayPurchase).filter_by(token_hash=play._hash(token)).all() 132 133 134 def test_the_play_build_can_sell_only_when_the_server_can_check_purchases( 135 client, monkeypatch, tmp_path): 136 monkeypatch.setattr(settings, "caption_api_key", "dg_test") 137 monkeypatch.setattr(settings, "stripe_secret_key", "") 138 monkeypatch.setattr(settings, "play_service_account_file", "") 139 assert client.get("/api/captions").json()["play_available"] is False 140 141 key = tmp_path / "play-key.json" 142 monkeypatch.setattr(settings, "play_service_account_file", str(key)) 143 assert client.get("/api/captions").json()["play_available"] is False # no file yet 144 145 key.write_text("{}") 146 state = client.get("/api/captions").json() 147 # The Play build does not need Stripe, and the web build does not need Google. 148 assert state["play_available"] is True 149 assert state["available"] is False 150 151 monkeypatch.setattr(settings, "caption_api_key", "") 152 assert client.get("/api/captions").json()["play_available"] is False 153 154 155 def assert_only_the_play_sales_stop(client, google): 156 """The Play build sells nothing. The GitHub build still sells with Stripe.""" 157 token = new_token() 158 google.buy(token, account_id(client)) 159 state = client.get("/api/captions") 160 assert state.status_code == 200 161 assert (state.json()["play_available"], state.json()["available"]) == (False, True) 162 r = post_play(client, token) 163 assert r.status_code == 503 and r.json()["detail"] == play.NOT_SET_UP 164 assert google.gets == [] 165 166 167 def test_a_key_folder_that_the_server_cannot_open_stops_only_the_play_sales( 168 client, google, monkeypatch, tmp_path): 169 # The service user cannot open the folder of the key, so stat of the key 170 # fails with EACCES. The server runs Python 3.11. There, Path.is_file 171 # raises PermissionError for this error, and each GET /api/captions 172 # answered 500. The tests can run on a newer Python, so this test makes 173 # Path.is_file act as on Python 3.11. The test does not make the key file: 174 # os.path.isfile gives False for each stat error, EACCES or ENOENT. 175 key = str(tmp_path / "locked" / "play-key.json") 176 is_file = pathlib.Path.is_file 177 178 def is_file_on_python_3_11(self, *args, **kwargs): 179 if str(self) == key: 180 raise PermissionError(errno.EACCES, "Permission denied", key) 181 return is_file(self, *args, **kwargs) 182 183 monkeypatch.setattr(pathlib.Path, "is_file", is_file_on_python_3_11) 184 monkeypatch.setattr(settings, "play_service_account_file", key) 185 monkeypatch.setattr(settings, "caption_api_key", "dg_test") 186 assert_only_the_play_sales_stop(client, google) 187 main.housekeeping() # The hourly pass does not fail. 188 189 190 def test_a_key_file_that_the_server_cannot_read_stops_only_the_play_sales( 191 client, google, monkeypatch): 192 # The key is there, but the service user cannot read it. For example, root 193 # owns it with mode 600. Then the server cannot check a purchase, so the 194 # Play build must not sell. 195 key = settings.play_service_account_file 196 access = os.access 197 198 def access_of_the_service_user(path, mode, *args, **kwargs): 199 if os.fspath(path) == key and mode & os.R_OK: 200 return False 201 return access(path, mode, *args, **kwargs) 202 203 monkeypatch.setattr(os, "access", access_of_the_service_user) 204 monkeypatch.setattr(settings, "caption_api_key", "dg_test") 205 assert_only_the_play_sales_stop(client, google) 206 207 208 # --- POST /api/captions/play-purchase ----------------------------------------- 209 210 def test_no_play_key_is_503_and_google_is_not_asked(client, google, monkeypatch, tmp_path): 211 token = new_token() 212 google.buy(token, account_id(client)) 213 monkeypatch.setattr(settings, "play_service_account_file", "") 214 r = post_play(client, token) 215 assert r.status_code == 503 and r.json()["detail"] == play.NOT_SET_UP 216 monkeypatch.setattr(settings, "play_service_account_file", str(tmp_path / "missing.json")) 217 assert post_play(client, token).status_code == 503 218 assert google.gets == [] and left(client) == 0 219 220 221 def test_an_unknown_pack_is_404(client, google): 222 token = new_token() 223 google.buy(token, account_id(client)) 224 assert post_play(client, token, product="pack10").status_code == 404 225 assert google.gets == [] 226 227 228 def test_a_paid_purchase_gives_its_hours_and_is_consumed(client, google): 229 token = new_token() 230 purchase = google.buy(token, account_id(client)) 231 r = post_play(client, token) 232 assert r.status_code == 200 and r.json() == {"seconds_left": 20 * HOUR} 233 assert google.consumed == [token] 234 assert left(client) == 20 * HOUR 235 [row] = ledger(token) 236 assert (row.account_id, row.product_id, row.quantity, row.seconds, row.order_id) == ( 237 account_id(client), "captions20", 1, 20 * HOUR, purchase["orderId"]) 238 239 240 def test_the_same_purchase_twice_gives_its_hours_once(client, google): 241 token = new_token() 242 google.buy(token, account_id(client)) 243 assert post_play(client, token).status_code == 200 244 r = post_play(client, token) # The app sends it again after a lost answer. 245 assert r.status_code == 200 and r.json() == {"seconds_left": 20 * HOUR} 246 assert google.consumed == [token] 247 assert len(ledger(token)) == 1 248 249 250 def test_a_pending_purchase_gives_nothing_until_it_is_paid(client, google): 251 token = new_token() 252 google.buy(token, account_id(client), state=2) 253 r = post_play(client, token) 254 assert r.status_code == 409 and r.json()["detail"] == "The purchase is not complete yet." 255 assert left(client) == 0 and google.consumed == [] and ledger(token) == [] 256 257 google.purchases[token]["purchaseState"] = 0 258 assert post_play(client, token).status_code == 200 259 assert left(client) == 20 * HOUR and google.consumed == [token] 260 261 262 def test_a_cancelled_purchase_gives_nothing(client, google): 263 token = new_token() 264 google.buy(token, account_id(client), state=1) 265 assert post_play(client, token).status_code == 409 266 assert left(client) == 0 and google.consumed == [] and ledger(token) == [] 267 268 269 def test_a_purchase_for_another_account_gives_nothing(client, second_client, google): 270 token, stranger = new_token(), new_token() 271 google.buy(token, account_id(client)) 272 google.buy(stranger, "acct_0000000000000000") # No such account on this server. 273 for sender, sent in ((second_client, token), (client, stranger)): 274 r = post_play(sender, sent) 275 assert r.status_code == 403 276 assert r.json()["detail"] == "This purchase belongs to another account." 277 assert left(client) == 0 and left(second_client) == 0 and google.consumed == [] 278 279 assert post_play(client, token).status_code == 200 280 assert left(client) == 20 * HOUR and left(second_client) == 0 281 282 283 def test_a_purchase_without_an_account_id_goes_to_the_caller_once(client, second_client, google): 284 token = new_token() 285 google.buy(token, None) 286 assert post_play(client, token).status_code == 200 287 r = post_play(second_client, token) 288 assert r.status_code == 200 and r.json() == {"seconds_left": 0} 289 assert left(client) == 20 * HOUR and left(second_client) == 0 290 assert google.consumed == [token] 291 292 293 def test_a_purchase_from_before_a_merge_goes_to_the_merged_account(client, second_client, google): 294 token = new_token() 295 old, survivor = account_id(client), account_id(second_client) 296 google.buy(token, old) 297 with SessionLocal() as s: 298 accounts.merge(s, s.get(Account, old), s.get(Account, survivor)) 299 s.commit() 300 assert post_play(second_client, token).status_code == 200 301 assert left(second_client) == 20 * HOUR 302 303 304 def test_a_failed_consume_keeps_the_hours_and_the_next_call_consumes(client, google): 305 token = new_token() 306 google.buy(token, account_id(client)) 307 google.fail_consume = 1 308 r = post_play(client, token) 309 assert r.status_code == 200 and r.json() == {"seconds_left": 20 * HOUR} 310 assert google.consumed == [] 311 312 assert post_play(client, token).status_code == 200 313 assert google.consumed == [token] and left(client) == 20 * HOUR 314 315 316 def test_a_failed_credit_leaves_no_row_and_no_consume(client, google, monkeypatch): 317 # The ledger row and the hours go in one commit, and the consume comes 318 # after that commit. So when the database stops between the row and the 319 # hours, Google keeps the purchase open, and the next send gives the hours. 320 token = new_token() 321 google.buy(token, account_id(client)) 322 add = captions.add 323 324 def database_stops(*args): 325 raise RuntimeError("the database stopped") 326 327 monkeypatch.setattr(captions, "add", database_stops) 328 with pytest.raises(RuntimeError): 329 post_play(client, token) 330 assert ledger(token) == [] and google.consumed == [] and google.early == [] 331 332 monkeypatch.setattr(captions, "add", add) 333 assert post_play(client, token).status_code == 200 334 assert left(client) == 20 * HOUR and google.consumed == [token] 335 336 337 def test_no_answer_from_google_gives_nothing(client, google): 338 token = new_token() 339 google.buy(token, account_id(client)) 340 google.down = True 341 r = post_play(client, token) 342 assert r.status_code == 502 and r.json()["detail"] == play.NO_ANSWER 343 assert left(client) == 0 and ledger(token) == [] 344 345 346 def test_a_token_that_google_does_not_know_is_400(client, google): 347 r = post_play(client, new_token()) 348 assert r.status_code == 400 and r.json()["detail"] == UNKNOWN 349 assert left(client) == 0 350 351 352 def test_a_purchase_for_another_pack_is_400(client, google): 353 token = new_token() 354 google.buy(token, account_id(client), product="captions80") 355 r = post_play(client, token, product="captions20") 356 assert r.status_code == 400 and r.json()["detail"] == UNKNOWN 357 assert left(client) == 0 and google.consumed == [] and ledger(token) == [] 358 359 360 def test_the_quantity_multiplies_the_hours(client, google): 361 token = new_token() 362 google.buy(token, account_id(client), quantity=3) 363 assert post_play(client, token).json() == {"seconds_left": 60 * HOUR} 364 365 366 def test_the_log_never_holds_the_token(client, google, caplog): 367 caplog.set_level(logging.DEBUG) 368 token, lost = new_token(), new_token() 369 google.buy(token, account_id(client)) 370 google.buy(lost, account_id(client))["consumptionState"] = 1 371 google.fail_consume = 1 372 assert post_play(client, token).status_code == 200 # The hours, and a failed consume. 373 assert post_play(client, token).status_code == 200 # The consume. 374 assert post_play(client, lost).status_code == 409 # Consumed, but not in the ledger. 375 assert "play purchase" in caplog.text and "could not consume" in caplog.text 376 assert token not in caplog.text and lost not in caplog.text 377 378 379 def test_a_consumed_purchase_without_a_row_gives_nothing(client, google): 380 token = new_token() 381 google.buy(token, account_id(client))["consumptionState"] = 1 382 r = post_play(client, token) 383 assert r.status_code == 409 and r.json()["detail"] == "This purchase was used already." 384 assert left(client) == 0 and google.consumed == [] and ledger(token) == [] 385 386 387 def test_a_test_purchase_is_marked_in_the_ledger(client, google): 388 tested, paid = new_token(), new_token() 389 google.buy(tested, account_id(client), test=True) 390 google.buy(paid, account_id(client)) 391 assert post_play(client, tested).status_code == 200 392 assert post_play(client, paid).status_code == 200 393 assert [row.test for row in ledger(tested)] == [True] 394 assert [row.test for row in ledger(paid)] == [False] 395 396 397 def test_a_bad_body_is_422(client, google): 398 token = new_token() 399 google.buy(token, account_id(client)) 400 assert post_play(client, "").status_code == 422 401 assert post_play(client, "x" * 4097).status_code == 422 402 assert google.gets == [] 403 404 405 # --- the HTTP layer ----------------------------------------------------------- 406 407 class FakeResponse: 408 def __init__(self, status_code: int, body: dict | None = None): 409 self.status_code = status_code 410 self._body = body or {} 411 412 def json(self) -> dict: 413 return self._body 414 415 416 class FakeHttp: 417 """Stands in for the signed session: records each request and gives the next answer.""" 418 419 def __init__(self): 420 self.answers: list = [] 421 self.seen: list[tuple[str, str]] = [] 422 423 def __enter__(self): 424 return self 425 426 def __exit__(self, *exc): 427 return False 428 429 def request(self, method: str, url: str, timeout: float | None = None): 430 assert timeout # A request without a timeout can hang a worker. 431 self.seen.append((method, url)) 432 answer = self.answers.pop(0) 433 if isinstance(answer, Exception): 434 raise answer 435 return answer 436 437 438 @pytest.fixture 439 def http(monkeypatch): 440 fake = FakeHttp() 441 monkeypatch.setattr(play, "_session", lambda: fake) 442 return fake 443 444 445 def test_get_asks_google_for_the_pack_and_the_token(http): 446 http.answers = [FakeResponse(200, {"purchaseState": 0})] 447 assert play.get("captions20", "abc.DEF-_1") == {"purchaseState": 0} 448 url = ("https://androidpublisher.googleapis.com/androidpublisher/v3/applications/" 449 "com.honjimaku.subrep/purchases/products/captions20/tokens/abc.DEF-_1") 450 assert http.seen == [("GET", url)] 451 452 453 def test_consume_posts_to_the_consume_address(http): 454 http.answers = [FakeResponse(204)] 455 play.consume("captions80", "abc") 456 [(method, url)] = http.seen 457 assert method == "POST" 458 assert url.endswith("/com.honjimaku.subrep/purchases/products/captions80/tokens/abc:consume") 459 460 461 @pytest.mark.parametrize("google_status, status", [ 462 (400, 400), (404, 400), (410, 400), (401, 503), (403, 503), (429, 502), (500, 502), (503, 502), 463 ]) 464 def test_google_errors_become_clear_answers(http, google_status, status): 465 http.answers = [FakeResponse(google_status)] 466 with pytest.raises(play.PlayError) as caught: 467 play.get("captions20", "abc") 468 assert caught.value.status == status 469 470 471 @pytest.mark.parametrize("error", [ 472 requests.ConnectionError("no route to host"), 473 requests.Timeout("read timed out"), 474 google_errors.TransportError("no route to the token server"), 475 # google-auth raises this after its own retries, when the Google token 476 # server answers 408, 429, 500, 503 or 504. 477 google_errors.RefreshError("temporarily_unavailable", retryable=True), 478 ]) 479 def test_no_connection_to_google_is_502(http, error, caplog): 480 http.answers = [error] 481 with pytest.raises(play.PlayError) as caught: 482 play.get("captions20", "abc") 483 assert (caught.value.status, caught.value.detail) == (502, play.NO_ANSWER) 484 # The key can be good. So the log must not tell the operator that Google refused it. 485 assert [r for r in caplog.records if r.levelno >= logging.ERROR] == [] 486 487 488 def test_a_key_that_google_refuses_is_503(http): 489 http.answers = [google_errors.RefreshError("invalid_grant: account not found")] 490 with pytest.raises(play.PlayError) as caught: 491 play.get("captions20", "abc") 492 assert (caught.value.status, caught.value.detail) == (503, play.NOT_SET_UP) 493 494 495 def test_the_key_file_gives_credentials_for_the_play_api(tmp_path): 496 from cryptography.hazmat.primitives import serialization 497 from cryptography.hazmat.primitives.asymmetric import rsa 498 499 pem = rsa.generate_private_key(public_exponent=65537, key_size=2048).private_bytes( 500 serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, 501 serialization.NoEncryption()).decode() 502 key = tmp_path / "play-key.json" 503 key.write_text(json.dumps({ 504 "type": "service_account", "project_id": "subrep-test", "private_key_id": "k1", 505 "private_key": pem, "client_email": "play@subrep-test.iam.gserviceaccount.com", 506 "client_id": "1", "token_uri": "https://oauth2.googleapis.com/token", 507 })) 508 credentials = play._credentials(str(key)) 509 assert credentials.scopes == ["https://www.googleapis.com/auth/androidpublisher"] 510 assert credentials.service_account_email == "play@subrep-test.iam.gserviceaccount.com" 511 512 513 @pytest.mark.parametrize("content", ["{}", "not json", None]) 514 def test_a_bad_key_file_is_503(tmp_path, content): 515 key = tmp_path / "play-key.json" 516 if content is not None: 517 key.write_text(content) 518 with pytest.raises(play.PlayError) as caught: 519 play._credentials(str(key)) 520 assert caught.value.status == 503 521 522 523 def test_the_voided_list_reads_every_page(http): 524 http.answers = [ 525 FakeResponse(200, {"voidedPurchases": [{"purchaseToken": "a", "orderId": "GPA.1"}], 526 "tokenPagination": {"nextPageToken": "p2"}}), 527 FakeResponse(200, {"voidedPurchases": [{"purchaseToken": "b", "orderId": "GPA.2"}]}), 528 FakeResponse(200, {}), 529 ] 530 assert play.voided() == ["a", "b"] 531 assert play.voided() == [] # No refunds in the last 30 days. 532 url = ("https://androidpublisher.googleapis.com/androidpublisher/v3/applications/" 533 "com.honjimaku.subrep/purchases/voidedpurchases") 534 assert http.seen == [("GET", url), ("GET", url + "?token=p2"), ("GET", url)] 535 536 537 # --- refunds ------------------------------------------------------------------ 538 # These tests use SessionLocal and no TestClient: each TestClient starts a pass 539 # of the housekeeping in the background. 540 541 def paid(product: str = "captions20", account: str | None = None) -> dict: 542 """A paid purchase that nobody consumed yet, as Google returns it.""" 543 purchase = {"purchaseState": 0, "consumptionState": 0, "productId": product} 544 if account is not None: 545 purchase["obfuscatedExternalAccountId"] = account 546 return purchase 547 548 549 def test_a_refund_takes_back_the_hours_once(): 550 token = new_token() 551 with SessionLocal() as s: 552 me = accounts.new_account(s) 553 captions.add(s, me.id, 5 * HOUR) # Hours from an earlier purchase. 554 s.commit() 555 assert play.credit(s, me, "captions20", token, paid(account=me.id)) 556 assert play.take_back(s, [token]) == 1 557 assert play.take_back(s, [token]) == 0 558 assert captions.seconds_left(s, me.id) == 5 * HOUR 559 [row] = s.query(play.PlayPurchase).filter_by(token_hash=play._hash(token)).all() 560 assert row.voided_at is not None 561 562 563 def test_two_passes_at_the_same_time_take_the_hours_back_once(monkeypatch): 564 token = new_token() 565 with SessionLocal() as s, SessionLocal() as other: 566 me = accounts.new_account(s) 567 captions.add(s, me.id, 5 * HOUR) 568 s.commit() 569 assert play.credit(s, me, "captions20", token, paid(account=me.id)) 570 # The other pass runs after this pass found the row, and before it 571 # marks the row. 572 other_pass = [] 573 execute = s.execute 574 575 def execute_after_the_other_pass(*args, **kwargs): 576 if not other_pass: 577 other_pass.append(play.take_back(other, [token])) 578 return execute(*args, **kwargs) 579 580 monkeypatch.setattr(s, "execute", execute_after_the_other_pass) 581 assert play.take_back(s, [token]) == 0 582 assert other_pass == [1] 583 me_id = me.id 584 with SessionLocal() as s: 585 assert captions.seconds_left(s, me_id) == 5 * HOUR 586 587 588 def test_a_refund_after_a_merge_takes_the_hours_from_the_survivor(): 589 token = new_token() 590 with SessionLocal() as s: 591 old, survivor = accounts.new_account(s), accounts.new_account(s) 592 s.commit() 593 assert play.credit(s, old, "captions20", token, paid(account=old.id)) 594 accounts.merge(s, old, survivor) 595 s.commit() 596 assert captions.seconds_left(s, survivor.id) == 20 * HOUR 597 assert play.take_back(s, [token]) == 1 598 assert captions.seconds_left(s, survivor.id) == 0 599 assert captions.seconds_left(s, old.id) == 0 600 601 602 def test_housekeeping_takes_back_refunds_only_when_play_is_set_up(google, monkeypatch): 603 token = new_token() 604 with SessionLocal() as s: 605 me = accounts.new_account(s) 606 s.commit() 607 assert play.credit(s, me, "captions20", token, paid(account=me.id)) 608 me_id = me.id 609 google.voided_tokens = [token] 610 asked = [] 611 612 def spy(): 613 asked.append(True) 614 return google.voided() 615 616 monkeypatch.setattr(play, "voided", spy) 617 key = settings.play_service_account_file 618 monkeypatch.setattr(settings, "play_service_account_file", "") 619 main.housekeeping() 620 assert asked == [] 621 622 monkeypatch.setattr(settings, "play_service_account_file", key) 623 main.housekeeping() 624 assert asked == [True] 625 with SessionLocal() as s: 626 assert captions.seconds_left(s, me_id) == 0 627 628 629 def test_a_google_failure_does_not_stop_housekeeping(google, monkeypatch, caplog): 630 def down(): 631 raise play.PlayError(502, play.NO_ANSWER) 632 633 monkeypatch.setattr(play, "voided", down) 634 main.housekeeping() 635 warnings = [record for record in caplog.records 636 if record.name == "backend.play" and record.levelno == logging.WARNING] 637 assert len(warnings) == 1 and play.NO_ANSWER in warnings[0].getMessage() 638 639 640 # --- properties --------------------------------------------------------------- 641 642 @hsettings(max_examples=50, deadline=None) 643 @given(buys=st.lists(st.tuples(st.sampled_from(sorted(captions.HOURS)), st.integers(1, 3), 644 st.sampled_from(["me", "other", "none"])), 645 min_size=1, max_size=4), 646 calls=st.lists(st.tuples(st.integers(0, 3), st.sampled_from([0, 1, 2]), 647 st.sampled_from([0, 1])), 648 max_size=25)) 649 def test_a_token_gives_its_hours_once_and_only_when_paid_for_this_account(buys, calls): 650 run = secrets.token_hex(6) # The database lives across examples. 651 with SessionLocal() as s: 652 me, other = accounts.new_account(s), accounts.new_account(s) 653 s.commit() 654 ids = {"me": me.id, "other": other.id, "none": None} 655 expected, added = {}, {} 656 for i, state, consumed in calls: 657 if i >= len(buys): 658 continue 659 product, quantity, owner = buys[i] 660 token = f"{run}.{i}" 661 google = {"purchaseState": state, "consumptionState": consumed, 662 "productId": product, "quantity": quantity} 663 if ids[owner]: 664 google["obfuscatedExternalAccountId"] = ids[owner] 665 try: 666 if play.credit(s, me, product, token, google): 667 added[token] = added.get(token, 0) + 1 668 except play.PlayError: 669 pass 670 # Hours come only from a call that saw the purchase paid and not consumed. 671 if state == 0 and consumed == 0 and owner != "other": 672 expected[token] = captions.HOURS[product] * HOUR * quantity 673 assert captions.seconds_left(s, ids["me"]) == sum(expected.values()) 674 assert captions.seconds_left(s, ids["other"]) == 0 675 assert added == {token: 1 for token in expected} 676 assert s.query(play.PlayPurchase).filter_by(account_id=ids["me"]).count() == len(expected) 677 678 679 @hsettings(max_examples=50, deadline=None) 680 @given(bought=st.lists(st.sampled_from(sorted(captions.HOURS)), min_size=1, max_size=4), 681 spent=st.integers(0, 300 * HOUR), 682 voids=st.lists(st.lists(st.integers(0, 5), max_size=6), max_size=4)) 683 def test_a_refund_takes_back_the_hours_once_and_never_below_zero(bought, spent, voids): 684 run = secrets.token_hex(6) # The database lives across examples. 685 with SessionLocal() as s: 686 me = accounts.new_account(s) 687 s.commit() 688 for i, product in enumerate(bought): 689 assert play.credit(s, me, product, f"{run}.{i}", paid(product, me.id)) 690 total = sum(captions.HOURS[product] * HOUR for product in bought) 691 used = min(spent, total) 692 assert captions.spend(s, me.id, used) 693 voided: set[int] = set() 694 taken = 0 695 for indexes in voids: # An index past the purchases is a token that this server never saw. 696 taken += play.take_back(s, [f"{run}.{i}" for i in indexes]) 697 voided |= {i for i in indexes if i < len(bought)} 698 back = sum(captions.HOURS[bought[i]] * HOUR for i in voided) 699 assert captions.seconds_left(s, me.id) == max(0, total - used - back) 700 assert taken == len(voided) 701 assert s.query(play.PlayPurchase).filter( 702 play.PlayPurchase.account_id == me.id, 703 play.PlayPurchase.voided_at.is_not(None)).count() == len(voided) 704 705 706 @given(st.text(min_size=1)) 707 def test_any_token_is_one_path_segment_of_the_address(token): 708 url = play._url("captions20", token) 709 assert "?" not in url and "#" not in url 710 head, tail = url.split("/tokens/", 1) 711 assert head.endswith("/com.honjimaku.subrep/purchases/products/captions20") 712 assert "/" not in tail 713 assert unquote(tail) == token