From 55314ef18c0d14231a8fb429755fec04af010a4a Mon Sep 17 00:00:00 2001 From: Xavier Thompson <xavier.thompson@nexedi.com> Date: Mon, 24 Jul 2023 15:08:53 +0200 Subject: [PATCH] slapos/testing: Add partition info utilities --- slapos/testing/testcase.py | 33 +++++++++++++++++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/slapos/testing/testcase.py b/slapos/testing/testcase.py index a5726bb6e..078aa3f09 100644 --- a/slapos/testing/testcase.py +++ b/slapos/testing/testcase.py @@ -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) @@ -410,6 +416,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 +653,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 -- 2.30.9