Commit 233394db authored by Xavier Thompson's avatar Xavier Thompson

Improve bits of testing framework and standalone

See merge request !560
parents 93ccc013 be20c700
......@@ -328,7 +328,7 @@ class SlapformatDefinitionWriter(ConfigWriter):
textwrap.dedent(
"""
[partition_{i}]
address = {ipv6_single_cidr} {ipv4_cidr}
address = {ipv4_cidr} {ipv6_single_cidr}
{ipv6_range_config_line}
pathname = {partition_base_name}{i}
user = {user}
......@@ -654,7 +654,6 @@ class StandaloneSlapOS(object):
unknown_partition_set.add(path)
# create partitions and configure computer
partition_list = []
for i in range(partition_count):
partition_reference = '%s%s' % (partition_base_name, i)
......@@ -663,25 +662,6 @@ class StandaloneSlapOS(object):
if not (os.path.exists(partition_path)):
os.mkdir(partition_path)
os.chmod(partition_path, 0o750)
ipv6_addr, ipv6_range = self._getPartitionIpv6(i)
partition_list.append({
'address_list': [
{
'addr': ipv4_address,
'netmask': NETMASK_IPV4_FULL
},
{
'addr': ipv6_addr,
'netmask': NETMASK_IPV6_FULL
}
],
'ipv6_range' : ipv6_range,
'path': partition_path,
'reference': partition_reference,
'tap': {
'name': partition_reference
},
})
if unknown_partition_set:
# sanity check that we are not removing partitions in use
......@@ -699,16 +679,6 @@ class StandaloneSlapOS(object):
"Cannot reformat to remove busy partition at {part_id}".format(
**locals()))
self.computer.updateConfiguration(
dumps({
'address': ipv4_address,
'netmask': NETMASK_IPV4_FULL,
'partition_list': partition_list,
'reference': self._computer_id,
'instance_root': self._instance_root,
'software_root': self._software_root
}))
for part in unknown_partition_set:
self._logger.debug(
"removing partition no longer part of format spec %s", part)
......
......@@ -32,21 +32,27 @@ import glob
import logging
import os
import shutil
import sqlite3
import unittest
import warnings
from six.moves.urllib.parse import urlparse
from netaddr import valid_ipv6
from .utils import getPortFromPath
from .utils import ManagedResource
from ..slap.standalone import StandaloneSlapOS
from ..slap.standalone import SlapOSNodeCommandError
from ..slap.standalone import PathTooDeepError
from ..util import mkdir_p
from ..slap import ComputerPartition
from .check_software import checkSoftware
from ..proxy.db_version import DB_VERSION
try:
from typing import Iterable, Tuple, Callable, Type, Dict, List, Optional, TypeVar
ManagedResourceType = TypeVar("ManagedResourceType", bound=ManagedResource)
......@@ -385,8 +391,7 @@ class SlapOSInstanceTestCase(unittest.TestCase):
max_retry=cls.instance_max_retry, debug=cls._debug)
@classmethod
def _setUpClass(cls):
cls.slap.start()
def formatPartitions(cls):
cls.logger.debug(
"Formatting to remove old partitions XXX should not be needed because we delete ..."
)
......@@ -396,12 +401,20 @@ class SlapOSInstanceTestCase(unittest.TestCase):
cls.partition_count, cls._ipv4_address, cls._ipv6_address,
getattr(cls, '__partition_reference__', '{}-'.format(cls.__name__)))
@classmethod
def _setUpClass(cls):
cls.slap.start()
# (re)format partitions
cls.formatPartitions()
# request
cls.requestDefaultInstance()
# slapos node instance
cls.logger.debug("Waiting for instance")
cls.waitForInstance()
# expose some class attributes so that tests can use them:
# the main ComputerPartition instance, to use getInstanceParameterDict
cls.computer_partition = cls.requestDefaultInstance()
......@@ -410,6 +423,10 @@ class SlapOSInstanceTestCase(unittest.TestCase):
cls.computer_partition_root_path = os.path.join(
cls.slap._instance_root, cls.computer_partition.getId())
# the ipv6 of the instance
cls.computer_partition_ipv6_address = cls.getPartitionIPv6(
cls.computer_partition.getId())
@classmethod
@contextlib.contextmanager
def _snapshotManager(cls, snapshot_name):
......@@ -643,3 +660,26 @@ class SlapOSInstanceTestCase(unittest.TestCase):
partition_reference=cls.default_partition_reference,
partition_parameter_kw=cls._instance_parameter_dict,
state=state)
@classmethod
def getPartitionId(cls, instance_name):
query = "SELECT reference FROM partition%s WHERE partition_reference=?" % DB_VERSION
with sqlite3.connect(os.path.join(
cls._base_directory,
'var/proxy.db',
)) as db:
return db.execute(query, (instance_name,)).fetchall()[0][0]
@classmethod
def getPartitionIPv6(cls, partition_id):
query = "SELECT address FROM partition_network%s WHERE partition_reference=?" % DB_VERSION
with sqlite3.connect(os.path.join(
cls._base_directory,
'var/proxy.db',
)) as db:
rows = db.execute(query, (partition_id,)).fetchall()
# do not assume the partition's IPv6 address is the second one,
# instead find the first address that is IPv6
for (address,) in rows:
if valid_ipv6(address):
return address
......@@ -136,14 +136,15 @@ class TestSlapOSStandaloneSetup(unittest.TestCase):
for i, resource in enumerate(resource_list):
resource_prefixlen = int(resource['ipv6_range']['network'].split('/')[1])
self.assertEqual(resource_prefixlen, prefixlen + 16)
self.assertTrue(netaddr.valid_ipv6(resource['address_list'][0]['addr']))
partition_ipv6 = resource['address_list'][1]['addr']
self.assertTrue(netaddr.valid_ipv6(partition_ipv6))
for other_resource in resource_list[i + 1:]:
self.assertNotEqual(
resource['ipv6_range']['addr'],
other_resource['ipv6_range']['addr'])
self.assertNotEqual(
resource['address_list'][0]['addr'],
other_resource['address_list'][0]['addr'])
partition_ipv6,
other_resource['address_list'][1]['addr'])
def test_format_ipv6_small_range(self):
standalone = self.setupSimpleStandalone()
......@@ -154,14 +155,12 @@ class TestSlapOSStandaloneSetup(unittest.TestCase):
resource_list = self.getJsonResourceList(standalone)
for i, resource in enumerate(resource_list):
self.assertFalse(resource['ipv6_range'])
self.assertTrue(netaddr.valid_ipv6(resource['address_list'][0]['addr']))
partition_ipv6 = resource['address_list'][1]['addr']
self.assertTrue(netaddr.valid_ipv6(partition_ipv6))
for other_resource in resource_list[i + 1:]:
self.assertNotEqual(
resource['address_list'][0]['addr'],
other_resource['address_list'][0]['addr'])
self.assertNotEqual(
resource['address_list'][0]['addr'],
addr0)
other_partition_ipv6 = other_resource['address_list'][1]['addr']
self.assertNotEqual(partition_ipv6, other_partition_ipv6)
self.assertNotEqual(partition_ipv6, addr0)
def test_format_ipv6_very_small_range(self):
standalone = self.setupSimpleStandalone()
......@@ -172,21 +171,17 @@ class TestSlapOSStandaloneSetup(unittest.TestCase):
resource_list = self.getJsonResourceList(standalone)
for i, resource in enumerate(resource_list):
self.assertFalse(resource['ipv6_range'])
self.assertTrue(netaddr.valid_ipv6(resource['address_list'][0]['addr']))
partition_ipv6 = resource['address_list'][1]['addr']
self.assertTrue(netaddr.valid_ipv6(partition_ipv6))
for j, other_resource in enumerate(resource_list[i + 1:]):
self.assertNotEqual(
resource['address_list'][0]['addr'],
addr0)
other_partition_ipv6 = other_resource['address_list'][1]['addr']
self.assertNotEqual(partition_ipv6, addr0)
if j % 2 == 1:
self.assertEqual(
resource['address_list'][0]['addr'],
other_resource['address_list'][0]['addr'])
self.assertEqual(partition_ipv6, other_partition_ipv6)
else:
self.assertNotEqual(
resource['address_list'][0]['addr'],
other_resource['address_list'][0]['addr'])
self.assertNotEqual(partition_ipv6, other_partition_ipv6)
def test_format_ipv6_slapsh_128_range(self):
def test_format_ipv6_slash_128_range(self):
standalone = self.setupSimpleStandalone()
prefixlen = 128
slapos_fake_ipv6_range = '%s/%d' % (SLAPOS_TEST_IPV6, prefixlen)
......@@ -194,10 +189,9 @@ class TestSlapOSStandaloneSetup(unittest.TestCase):
resource_list = self.getJsonResourceList(standalone)
for i, resource in enumerate(resource_list):
self.assertFalse(resource['ipv6_range'])
self.assertTrue(netaddr.valid_ipv6(resource['address_list'][0]['addr']))
self.assertEqual(
resource['address_list'][0]['addr'],
SLAPOS_TEST_IPV6)
partition_ipv6 = resource['address_list'][1]['addr']
self.assertTrue(netaddr.valid_ipv6(partition_ipv6))
self.assertEqual(partition_ipv6, SLAPOS_TEST_IPV6)
def test_format_ipv6_no_range(self):
standalone = self.setupSimpleStandalone()
......@@ -205,11 +199,10 @@ class TestSlapOSStandaloneSetup(unittest.TestCase):
resource_list = self.getJsonResourceList(standalone)
for i, resource in enumerate(resource_list):
self.assertFalse(resource['ipv6_range'])
self.assertTrue(netaddr.valid_ipv6(resource['address_list'][0]['addr']))
partition_ipv6 = resource['address_list'][1]['addr']
self.assertTrue(netaddr.valid_ipv6(partition_ipv6))
for other_resource in resource_list[i + 1:]:
self.assertEqual(
resource['address_list'][0]['addr'],
SLAPOS_TEST_IPV6)
self.assertEqual(partition_ipv6, SLAPOS_TEST_IPV6)
def test_reformat_less_partitions(self):
standalone = self.setupSimpleStandalone()
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment