# Check if the file has been removed
self.assertFalse(os.path.exists(upload.path))
+ async def test_write_too_large_file(self):
+ """
+ Creates an upload of a certain size, but then tries to write more data
+ """
+ with self.db.transaction():
+ upload = self.backend.uploads.create("test.blob", size=20, user=self.user)
+
+ # Try to write more than 20 bytes
+ with self.assertRaises(OverflowError):
+ await upload.write(b"012345678901234567890123456789")
+
+ async def test_check_digest(self):
+ """
+ Creates an upload and checks if the digest matches
+ """
+ with self.db.transaction():
+ upload = self.backend.uploads.create("test.blob", size=20, user=self.user)
+
+ # Write some payload
+ await upload.write(b"01234567890123456789")
+
+ digest = bytes.fromhex("185c728c3fccb51875d74e21fec19f4cdfad8d5aa347a7a75c06473"
+ "af6f73835b5a00515a34f0e09725d5b1e0715ce43a1a57d966a92400efd215e45dd19c09c")
+
+ # Check the digest
+ self.assertTrue(await upload.check_digest("blake2b", digest))
+
if __name__ == "__main__":
unittest.main()