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
|
#!/usr/bin/python3
import os
import sys
import time
import libvirt
import unittest
class test_ipxe(unittest.TestCase):
def setUp(self):
self.arch = os.uname()[4]
self.conn = libvirt.open(None)
self.network = {}
self.domain = {}
def tearDown(self):
for (mode, domain) in self.domain.items():
domain.destroy()
for (mode, network) in self.network.items():
network.destroy()
self.conn.close()
def do_wait(self, stream, substr, timeout, verbose = False):
start = time.time()
log = b''
while time.time() - start < timeout:
data = stream.recv(128)
if type(data) is int and data == -2:
continue
if verbose:
print(data)
log += data
if substr in log:
secs = time.time() - start
print(f'{substr.decode()} ({secs:.1f}s) ... ', end = '', flush = True)
return
time.sleep(0.01)
self.fail(f'not found: {substr}')
def do_ipxe(self, mode):
name = f'test-ipxe-{mode}-{self.arch}'
network = self.conn.networkLookupByName(name)
if not network.isActive():
network.create()
self.network[mode] = network
domain = self.conn.lookupByName(name)
domain.create()
self.domain[mode] = domain
stream = self.conn.newStream()
domain.openConsole(None, stream, 0)
self.do_wait(stream, b'Start PXE', 20)
self.do_wait(stream, b'iPXE', 40)
self.do_wait(stream, b'hello world', 90)
def test_ipv4(self):
self.do_ipxe('ipv4')
def test_ipv6(self):
self.do_ipxe('ipv6')
if __name__ == '__main__':
unittest.main()
|
