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

1from urllib.parse import parse_qs, urlparse 

2 

3import pytest 

4from google.protobuf import empty_pb2 

5from sqlalchemy import select 

6 

7from couchers.crypto import b64decode, random_hex 

8from couchers.db import session_scope 

9from couchers.models import InitiatedUpload, Upload 

10from couchers.proto import media_pb2 

11from tests.fixtures.db import generate_user 

12from tests.fixtures.sessions import api_session, media_session 

13 

14 

15@pytest.fixture(autouse=True) 

16def _(testconfig): 

17 pass 

18 

19 

20def test_media_upload(db): 

21 user, token = generate_user() 

22 

23 media_bearer_token = random_hex(32) 

24 

25 with api_session(token) as api: 

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

27 

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

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

30 

31 response = media_pb2.UploadRequest.FromString(data) 

32 key = response.key 

33 

34 filename = random_hex(32) 

35 

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

37 

38 with session_scope() as session: 

39 # make sure it exists 

40 assert ( 

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

42 ) 

43 

44 with media_session(media_bearer_token) as media: 

45 res = media.UploadConfirmation(req) 

46 

47 with session_scope() as session: 

48 # make sure it exists 

49 assert ( 

50 session.execute( 

51 select(Upload) 

52 .where(Upload.key == key) 

53 .where(Upload.filename == filename) 

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

55 ).scalar_one_or_none() 

56 is not None 

57 ) 

58 

59 with session_scope() as session: 

60 # make sure it was deleted 

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