1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
|
#! /usr/bin/env python3
#
# Copyright (C) 2024 BitBake Contributors
#
# SPDX-License-Identifier: GPL-2.0-only
#
from . import create_server, create_client
import sys
import tempfile
import unittest
import socket
from pathlib import Path
THIS_DIR = Path(__file__).parent
BIN_DIR = THIS_DIR.parent.parent / "bin"
def server_prefunc(server, idx):
logging.basicConfig(level=logging.DEBUG, filename='bbprserv-%d.log' % idx, filemode='w',
format='%(levelname)s %(filename)s:%(lineno)d %(message)s')
server.logger.debug("Running server %d" % idx)
sys.stdout = open('bbprserv-stdout-%d.log' % idx, 'w')
sys.stderr = sys.stdout
class PRTestSetup(object):
server_index = 0
client_index = 0
def start_server(self, dbpath=None, upstream=None, read_only=False, prefunc=server_prefunc):
self.server_index += 1
if dbpath is None:
dbpath = self.make_dbpath()
def cleanup_server(server):
if server.process.exitcode is not None:
return
server.process.terminate()
server.process.join()
server = create_server(socket.gethostbyname("localhost") + ":0",
dbpath,
upstream=upstream,
read_only=read_only)
server.serve_as_process(prefunc=prefunc, args=(self.server_index))
self.addCleanup(cleanup_server, server)
return server
def make_dbpath(self):
return os.path.join(self.temp_dir.name, "prserv-test%d.sqlite3" % self.server_index)
def start_client(self, server_address):
def cleanup_client(client):
client.close()
client = create_client(server_address)
self.addCleanup(cleanup_client, client)
return client
def start_test_server(self):
self.server = self.start_server()
print("SERVER STARTED WITH ADDRESS: %s" % (self.server.address))
return self.server.address
def setUp(self):
self.temp_dir = tempfile.TemporaryDirectory(prefix='bb-prserv-test')
self.addCleanup(self.temp_dir.cleanup)
self.server_address = self.start_test_server()
self.client = self.start_client(self.server_address)
class PRCommonTests(PRTestSetup, unittest.TestCase):
def test_create_pr(self):
result = self.client.test_pr("dummy-1.0-r0", "core2-64", "51bf8189dbe9ea81fa6dd89608bf19380c437a9cf12f6c6239887801ba4ab4a5")
return self.assertEqual(result, None)
|
