Coverage for src/tests/test_media.py: 100%

1from urllib.parse import parse_qs, urlparse 

2 

3import pytest 

4from google.protobuf import empty_pb2 

5 

6from couchers.crypto import b64decode, random_hex 

7from couchers.db import session_scope 

8from couchers.models import InitiatedUpload, Upload 

9from couchers.sql import couchers_select as select 

10from proto import media_pb2 

11from tests.test_fixtures import api_session, db, generate_user, media_session, testconfig # noqa 

12 

13 

14@pytest.fixture(autouse=True) 

15def _(testconfig): 

16 pass 

17 

18 

19def test_media_upload(db): 

20 user, token = generate_user() 

21 

22 media_bearer_token = random_hex(32) 

23 

24 with api_session(token) as api: 

25 res = api.InitiateMediaUpload(empty_pb2.Empty()) 

26 

27 params = parse_qs(urlparse(res.upload_url).query) 

28 data = b64decode(params["data"][0]) 

29 

30 response = media_pb2.UploadRequest.FromString(data) 

31 key = response.key 

32 

33 filename = random_hex(32) 

34 

35 req = media_pb2.UploadConfirmationReq(key=key, filename=filename) 

36 

37 with session_scope() as session: 

38 # make sure it exists 

39 assert ( 

40 session.execute(select(InitiatedUpload).where(InitiatedUpload.key == key)).scalar_one_or_none() is not None 

41 ) 

42 

43 with media_session(media_bearer_token) as media: 

44 res = media.UploadConfirmation(req) 

45 

46 with session_scope() as session: 

47 # make sure it exists 

48 assert ( 

49 session.execute( 

50 select(Upload) 

51 .where(Upload.key == key) 

52 .where(Upload.filename == filename) 

53 .where(Upload.creator_user_id == user.id) 

54 ).scalar_one() 

55 is not None 

56 ) 

57 

58 with session_scope() as session: 

59 # make sure it was deleted 

60 assert not session.execute(select(InitiatedUpload)).scalar_one_or_none()