tests/upload.py

   1 #!/usr/bin/python3
   2
   3 import io
   4 import os
   5 import unittest
   6
   7 import test
   8
   9 from buildservice import uploads
  10 from buildservice import users
  11
  12 class UploadTestCase(test.TestCase):
  13         """
  14                 Tests everything around uploads
  15         """
  16         async def test_create_delete(self):
  17                 """
  18                         Tests whether uploads can be created and deleted
  19                 """
  20                 # Create the upload object
  21                 upload = self.backend.uploads.create("test.blob", size=0, owner=self.user)
  22
  23                 self.assertIsInstance(upload, uploads.Upload)
  24                 self.assertEqual(upload.filename, "test.blob")
  25
  26                 # Check if the upload file exists
  27                 self.assertTrue(os.path.exists(upload.path))
  28
  29                 # Delete the upload
  30                 await upload.delete()
  31
  32                 # Check if the file has been removed
  33                 self.assertFalse(os.path.exists(upload.path))
  34
  35         async def test_write_too_large_file(self):
  36                 """
  37                         Creates an upload of a certain size, but then tries to write more data
  38                 """
  39                 payload = io.BytesIO(b"012345678901234567890123456789")
  40
  41                 with self.db.transaction():
  42                         upload = await self.backend.uploads.create("test.blob", size=20, owner=self.user)
  43
  44                 # Try to write more than 20 bytes
  45                 with self.assertRaises(OverflowError):
  46                         await upload.copyfrom(payload)
  47
  48         async def test_check_digest(self):
  49                 """
  50                         Creates an upload and checks if the digest matches
  51                 """
  52                 payload = io.BytesIO(b"01234567890123456789")
  53
  54                 with self.db.transaction():
  55                         upload = await self.backend.uploads.create("test.blob", size=20, owner=self.user)
  56
  57                 # Write some payload
  58                 await upload.copyfrom(payload)
  59
  60                 digest = bytes.fromhex("185c728c3fccb51875d74e21fec19f4cdfad8d5aa347a7a75c06473"
  61                         "af6f73835b5a00515a34f0e09725d5b1e0715ce43a1a57d966a92400efd215e45dd19c09c")
  62
  63                 # Check the digest
  64                 self.assertTrue(await upload.check_digest("blake2b", digest))
  65
  66         async def test_quota(self):
  67                 """
  68                         Tries to create an upload that exceeds the quota
  69                 """
  70                 # Create an upload that is 200 MiB large
  71                 with self.db.transaction():
  72                         with self.assertRaises(users.QuotaExceededError):
  73                                 await self.backend.uploads.create(
  74                                         "test.blob",
  75                                         size=209715200,
  76                                         owner=self.user,
  77                                 )
  78
  79
  80 if __name__ == "__main__":
  81         unittest.main()