# The score should now be 1
self.assertEqual(build.score, 1)
+ async def test_run(self):
+ """
+ This test creates a build and tries to emulate a job being completed.
+ """
+ path = self.source_path("tests/data/beep-1.3-2.ip3.src.pfm")
+
+ # Create the build
+ with self.db.transaction():
+ build = await self._create_build(path)
+
+ # Assign all jobs to the default builder
+ for job in build.jobs:
+ job.assign(self.builder)
+
+ # This job should now be running
+ self.assertTrue(job.is_running())
+
+ # The builder should match
+ self.assertEqual(job.builder, self.builder)
+
+ path = self.source_path("tests/data/beep-1.3-2.ip3.x86_64.pfm")
+
+ # Upload a package
+ upload = await self._create_upload(path)
+
+ # Pretend that all jobs finished
+ for job in build.jobs:
+ await job.finished(success=True, packages=[upload])
+
if __name__ == "__main__":
unittest.main()
import configparser
import functools
import os
+import shutil
import socket
import tempfile
import unittest
from buildservice import database
from buildservice import misc
from buildservice import packages
+from buildservice import uploads
class TestCase(unittest.IsolatedAsyncioTestCase):
"""
self.assertIsInstance(build, builds.Build)
return build
+
+ async def _create_upload(self, path, filename=None, user=None, **kwargs):
+ """
+ Helper function to create an upload from path
+ """
+ if filename is None:
+ filename = os.path.basename(path)
+
+ if user is None:
+ user = self.user
+
+ # Allocate a new destination file
+ dst = self.backend.uploads.allocate_file()
+
+ # Open the source file
+ with open(path, "rb") as src:
+ # Copy the entire content
+ shutil.copyfileobj(src, dst)
+
+ # Close the destination file
+ dst.close()
+
+ # Create the upload object
+ upload = self.backend.uploads.create(filename, dst.name, user=user, **kwargs)
+
+ # Check if received the correct type
+ self.assertIsInstance(upload, uploads.Upload)
+
+ return upload