tests/test.py

   1 #!/usr/bin/python3
   2
   3 import configparser
   4 import functools
   5 import os
   6 import socket
   7 import tempfile
   8 import unittest
   9
  10 from pakfire._pakfire import Archive
  11
  12 from buildservice import Backend
  13 from buildservice import builds
  14 from buildservice import database
  15 from buildservice import misc
  16 from buildservice import packages
  17 from buildservice import uploads
  18
  19 class TestCase(unittest.IsolatedAsyncioTestCase):
  20         """
  21                 This is the base class for all tests
  22         """
  23         def source_path(self, path):
  24                 try:
  25                         source_path = os.environ["abs_top_srcdir"]
  26                 except KeyError as e:
  27                         raise ValueError("source path is not set") from e
  28
  29                 return os.path.join(source_path, path)
  30
  31         def _setup_database(self):
  32                 """
  33                         Creates a new database and imports the default schema
  34                 """
  35                 # Path to the schema
  36                 schema = self.source_path("src/database.sql")
  37
  38                 # Load the schema
  39                 with open(schema) as f:
  40                         schema = f.read()
  41
  42                 # Database Server & Name
  43                 host = os.environ.get("DB_HOST", socket.gethostname())
  44                 name = os.environ.get("DB_NAME", "pakfiretest")
  45
  46                 # Credentials
  47                 username = os.environ.get("DB_USER", "pakfiretest")
  48                 password = os.environ.get("DB_PASS", "pakfiretest")
  49
  50                 # Connect to the database
  51                 db = database.Connection(host, name, username, password)
  52
  53                 with db.transaction():
  54                         # Drop anything existing
  55                         db.execute("DROP SCHEMA public CASCADE")
  56
  57                         # Re-create the schema
  58                         db.execute("CREATE SCHEMA public")
  59                         db.execute(schema)
  60
  61                 # Return the credentials
  62                 return {
  63                         "name"     : name,
  64                         "hostname" : host,
  65                         "user"     : username,
  66                         "password" : password,
  67                 }
  68
  69         async def asyncSetUp(self):
  70                 # Create a new temporary directory
  71                 self.testdir = tempfile.TemporaryDirectory()
  72
  73                 # Create a configuration file
  74                 conf = configparser.ConfigParser()
  75
  76                 # Set the base path
  77                 conf["global"] = {
  78                         "basepath" : self.testdir.name,
  79                 }
  80
  81                 # Setup the database
  82                 conf["database"] = self._setup_database()
  83
  84                 with tempfile.NamedTemporaryFile("w") as f:
  85                         # Write configuration
  86                         conf.write(f)
  87
  88                         # Flush
  89                         f.flush()
  90
  91                         # Initialize the backend
  92                         self.backend = Backend(f.name, test=True)
  93
  94                 # Create handle to the database
  95                 self.db = self.backend.db
  96
  97                 # Create some default objects
  98                 with self.db.transaction():
  99                         await self._create_default_objects()
 100
 101         async def asyncTearDown(self):
 102                 # Dump any messages in the message queue
 103                 for message in self.backend.messages.queue:
 104                         print(message.message)
 105
 106                 # Dump a listing of all temporary files
 107                 with os.scandir(self.testdir.name) as listing:
 108                         for entry in sorted(listing, key=lambda e: e.path):
 109                                 print("  %s" % entry.path)
 110
 111                 # Removing any temporary files
 112                 self.testdir.cleanup()
 113
 114         async def _create_default_objects(self):
 115                 """
 116                         Creates some random objects that are created by default so
 117                         that we won't have to start from scratch each time...
 118                 """
 119                 # Create a builder
 120                 self.builder = self.backend.builders.create("test-builder01.ipfire.org")
 121
 122                 # Create a user
 123                 self.user = self.backend.users.create("tester", _attrs={
 124                         "cn"    : [b"Joe Tester"],
 125                         "email" : [b"joe.tester@ipfire.org"],
 126                 })
 127
 128                 # Set storage quota
 129                 self.user.storage_quota = 104857600 # 100 MiB
 130
 131                 # Create a distribution
 132                 self.distro = self.backend.distros.create("Default Test Distribution", "test", 1)
 133
 134                 # Enable this distribution for aarch64 & x86_64
 135                 self.distro.arches = ["aarch64", "x86_64"]
 136
 137                 # Create a repository
 138                 self.repo = await self.backend.repos.create(self.distro, "Default Test Repository")
 139
 140         async def _create_package(self, path):
 141                 """
 142                         Helper function to import a package from path
 143                 """
 144                 # Check if the file exists
 145                 self.assertTrue(os.path.exists(path))
 146
 147                 # Upload the file
 148                 upload = await self._create_upload(path)
 149
 150                 # Create the package
 151                 package = await self.backend.packages.create(upload, distro=self.distro)
 152
 153                 # Check if we received the correct type
 154                 self.assertIsInstance(package, packages.Package)
 155
 156                 return package
 157
 158         async def _create_build(self, path, repo=None, owner=None):
 159                 """
 160                         Helper function to import a build from path
 161                 """
 162                 # Use the default repository if none set
 163                 if repo is None:
 164                         repo = self.repo
 165
 166                 # Import the package first
 167                 package = await self._create_package(path)
 168
 169                 # Create the build
 170                 build = await self.backend.builds.create(repo, package, owner=owner)
 171
 172                 # Check if we received the correct type
 173                 self.assertIsInstance(build, builds.Build)
 174
 175                 return build
 176
 177         async def _create_upload(self, path, **kwargs):
 178                 """
 179                         Helper function to create an upload from path
 180                 """
 181                 # Create the upload object
 182                 upload = await self.backend.uploads.create_from_local(path, **kwargs)
 183
 184                 # Check if received the correct type
 185                 self.assertIsInstance(upload, uploads.Upload)
 186
 187                 return upload