StealthOS/subiquity
StealthOS
/
subiquity
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

mirror: introduce new primary entry format but do not use yet 

Signed-off-by: Olivier Gayot <olivier.gayot@canonical.com>
This commit is contained in:
Olivier Gayot 2023-01-29 18:26:25 +01:00
parent 23db4b597b
commit f7cd813254
2 changed files with 110 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

75
subiquity/models/mirror.py
View File

@ -18,6 +18,8 @@ import logging
from typing import Any, Dict, List, Optional, Set
from urllib import parse
import attr
from curtin.commands.apt_config import (
get_arch_mirrorconfig,
get_mirror,
@ -32,28 +34,67 @@ except ImportError:
log = logging.getLogger('subiquity.models.mirror')
DEFAULT_SUPPORTED_ARCHES_URI = "http://archive.ubuntu.com/ubuntu"
DEFAULT_PORTS_ARCHES_URI = "http://ports.ubuntu.com/ubuntu-ports"
DEFAULT_PRIMARY_SECTION = [
LEGACY_DEFAULT_PRIMARY_SECTION = [
{
"arches": PRIMARY_ARCHES,
"uri": "http://archive.ubuntu.com/ubuntu",
"uri": DEFAULT_SUPPORTED_ARCHES_URI,
}, {
"arches": ["default"],
"uri": "http://ports.ubuntu.com/ubuntu-ports",
"uri": DEFAULT_PORTS_ARCHES_URI,
},
]
DEFAULT = {
"preserve_sources_list": False,
}
class PrimarySection:
""" Helper to manage a primary autoinstall section. """
@attr.s(auto_attribs=True)
class PrimaryElement:
parent: "MirrorModel" = attr.ib(kw_only=True)
@attr.s(auto_attribs=True)
class PrimaryEntry(PrimaryElement):
# If the uri is None, it indicates a country-mirror that has not yet been
# resolved.
uri: Optional[str] = None
arches: Optional[List[str]] = None
@classmethod
def from_config(cls, config: Any, parent: "MirrorModel") -> "PrimaryEntry":
if config == "country-mirror":
return cls(parent=parent)
if config.get("uri", None) is None:
raise ValueError("uri is mandatory")
return cls(**config, parent=parent)
@property
def config(self) -> List[Dict[str, Any]]:
assert self.uri is not None
arches = []
if self.arches is None:
arches = [self.parent.architecture]
return [{"uri": self.uri, "arches": arches}]
def supports_arch(self, arch: str) -> bool:
""" Tells whether the mirror claims to support the architecture
specified. """
if self.arches is None:
return True
return arch in self.arches
class LegacyPrimarySection(PrimaryElement):
""" Helper to manage a apt->primary autoinstall section.
The format is the same as the format expected by curtin, no more, no less.
"""
def __init__(self, config: List[Any], *, parent: "MirrorModel") -> None:
self.parent = parent
self.config = config
super().__init__(parent=parent)
@property
def uri(self) -> str:
@ -72,8 +113,9 @@ class PrimarySection:
return self.uri == self.parent.default_mirror
@classmethod
def new_from_default(cls, parent: "MirrorModel") -> "PrimarySection":
return cls(copy.deepcopy(DEFAULT_PRIMARY_SECTION), parent=parent)
def new_from_default(cls, parent: "MirrorModel") -> "LegacyPrimarySection":
return cls(copy.deepcopy(LEGACY_DEFAULT_PRIMARY_SECTION),
parent=parent)
def countrify_uri(uri: str, cc: str) -> str:
@ -88,12 +130,12 @@ class MirrorModel(object):
def __init__(self):
self.config = copy.deepcopy(DEFAULT)
self.disabled_components: Set[str] = set()
self.primary_elected: Optional[PrimarySection] = None
self.primary_candidates: List[PrimarySection] = [
PrimarySection.new_from_default(parent=self),
self.primary_elected: Optional[LegacyPrimarySection] = None
self.primary_candidates: List[LegacyPrimarySection] = [
LegacyPrimarySection.new_from_default(parent=self),
]
self.primary_staged: Optional[PrimarySection] = None
self.primary_staged: Optional[LegacyPrimarySection] = None
self.architecture = get_architecture()
self.default_mirror = self.primary_candidates[0].uri
@ -104,7 +146,7 @@ class MirrorModel(object):
if "primary" in data:
# TODO support multiple candidates.
self.primary_candidates = [
PrimarySection(data.pop("primary"), parent=self)
LegacyPrimarySection(data.pop("primary"), parent=self)
]
merge_config(self.config, data)
@ -149,14 +191,15 @@ class MirrorModel(object):
def replace_primary_candidates(self, uris: List[str]) -> None:
self.primary_candidates.clear()
for uri in uris:
section = PrimarySection.new_from_default(parent=self)
section = LegacyPrimarySection.new_from_default(parent=self)
section.uri = uri
self.primary_candidates.append(section)
# NOTE: this is sometimes useful but it can be troublesome as well.
self.primary_staged = None
def assign_primary_elected(self, uri: str) -> None:
self.primary_elected = PrimarySection.new_from_default(parent=self)
self.primary_elected = \
LegacyPrimarySection.new_from_default(parent=self)
self.primary_elected.uri = uri
def wants_geoip(self) -> bool:

61
subiquity/models/tests/test_mirror.py
View File

@ -17,9 +17,10 @@ import unittest
from subiquity.models.mirror import (
countrify_uri,
DEFAULT_PRIMARY_SECTION,
LEGACY_DEFAULT_PRIMARY_SECTION,
MirrorModel,
PrimarySection,
LegacyPrimarySection,
PrimaryEntry,
)
@ -51,27 +52,67 @@ class TestCountrifyUrl(unittest.TestCase):
"http://us.ports.ubuntu.com/ubuntu-ports")
class TestPrimarySection(unittest.TestCase):
class TestPrimaryEntry(unittest.TestCase):
def test_initializer(self):
model = MirrorModel()
entry = PrimaryEntry(parent=model)
self.assertEqual(entry.parent, model)
self.assertIsNone(entry.uri, None)
self.assertIsNone(entry.arches, None)
entry = PrimaryEntry("http://mirror", ["amd64"], parent=model)
self.assertEqual(entry.parent, model)
self.assertEqual(entry.uri, "http://mirror")
self.assertEqual(entry.arches, ["amd64"])
entry = PrimaryEntry(uri="http://mirror", arches=[], parent=model)
self.assertEqual(entry.parent, model)
self.assertEqual(entry.uri, "http://mirror")
self.assertEqual(entry.arches, [])
def test_from_config(self):
model = MirrorModel()
entry = PrimaryEntry.from_config("country-mirror", parent=model)
self.assertEqual(entry, PrimaryEntry(parent=model))
with self.assertRaises(ValueError):
entry = PrimaryEntry.from_config({}, parent=model)
entry = PrimaryEntry.from_config(
{"uri": "http://mirror"}, parent=model)
self.assertEqual(entry, PrimaryEntry(
uri="http://mirror", parent=model))
entry = PrimaryEntry.from_config(
{"uri": "http://mirror", "arches": ["amd64"]}, parent=model)
self.assertEqual(entry, PrimaryEntry(
uri="http://mirror", arches=["amd64"], parent=model))
class TestLegacyPrimarySection(unittest.TestCase):
def setUp(self):
self.model = MirrorModel()
def test_initializer(self):
primary = PrimarySection([], parent=self.model)
primary = LegacyPrimarySection([], parent=self.model)
self.assertEqual(primary.config, [])
self.assertEqual(primary.parent, self.model)
def test_new_from_default(self):
primary = PrimarySection.new_from_default(parent=self.model)
self.assertEqual(primary.config, DEFAULT_PRIMARY_SECTION)
primary = LegacyPrimarySection.new_from_default(parent=self.model)
self.assertEqual(primary.config, LEGACY_DEFAULT_PRIMARY_SECTION)
def test_get_uri(self):
self.model.architecture = "amd64"
primary = PrimarySection([{"uri": "http://myurl", "arches": "amd64"}],
parent=self.model)
primary = LegacyPrimarySection(
[{"uri": "http://myurl", "arches": "amd64"}],
parent=self.model)
self.assertEqual(primary.uri, "http://myurl")
def test_set_uri(self):
primary = PrimarySection.new_from_default(parent=self.model)
primary = LegacyPrimarySection.new_from_default(parent=self.model)
primary.uri = "http://mymirror.invalid/"
self.assertEqual(primary.uri, "http://mymirror.invalid/")
@ -132,7 +173,7 @@ class TestMirrorModel(unittest.TestCase):
primary = [{"arches": "amd64", "uri": "http://mirror"}]
self.model.disabled_components = set(["non-free"])
self.model.primary_candidates = \
[PrimarySection(primary, parent=self.model)]
[LegacyPrimarySection(primary, parent=self.model)]
self.model.primary_elected = self.model.primary_candidates[0]
cfg = self.model.make_autoinstall()
self.assertEqual(cfg["disable_components"], ["non-free"])