Coverage for daklib/config.py: 66%
77 statements
« prev ^ index » next coverage.py v7.6.0, created at 2026-08-03 16:46 +0000
« prev ^ index » next coverage.py v7.6.0, created at 2026-08-03 16:46 +0000
1"""
2Config access class
4@contact: Debian FTPMaster <ftpmaster@debian.org>
5@copyright: 2008 Mark Hymers <mhy@debian.org>
6@license: GNU General Public License version 2 or later
7"""
9# This program is free software; you can redistribute it and/or modify
10# it under the terms of the GNU General Public License as published by
11# the Free Software Foundation; either version 2 of the License, or
12# (at your option) any later version.
14# This program is distributed in the hope that it will be useful,
15# but WITHOUT ANY WARRANTY; without even the implied warranty of
16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17# GNU General Public License for more details.
19# You should have received a copy of the GNU General Public License
20# along with this program; if not, write to the Free Software
21# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
23################################################################################
25# <NCommander> mhy, how about "Now with 20% more monty python references"
27################################################################################
29import getpass
30import grp
31import os
32import socket
33from typing import TYPE_CHECKING, Any
35import apt_pkg
37if TYPE_CHECKING:
38 from daklib.dbconn import Suite
40################################################################################
42default_config = "/etc/dak/dak.conf" #: default dak config, defines host properties
44# suppress some deprecation warnings in squeeze related to apt_pkg
45# module
46import warnings
48warnings.filterwarnings("ignore", ".*apt_pkg.* is deprecated.*", DeprecationWarning)
50################################################################################
53def which_conf_file() -> str:
54 return os.getenv("DAK_CONFIG", default_config)
57class Config:
58 """
59 A Config object is a singleton containing
60 information about the DAK configuration
61 """
63 __shared_state: dict[str, Any] = {}
65 def __init__(self, *args, **kwargs):
66 self.__dict__ = self.__shared_state
68 if not getattr(self, "initialised", False):
69 self.initialised = True
70 self._readconf()
71 self._setup_routines()
73 def _readconf(self):
74 apt_pkg.init()
76 self.Cnf = apt_pkg.Configuration()
78 apt_pkg.read_config_file_isc(self.Cnf, which_conf_file()) # type: ignore[attr-defined]
80 # Check whether our dak.conf was the real one or
81 # just a pointer to our main one
82 fqdn = socket.getfqdn()
83 conffile = self.Cnf.get("Config::" + fqdn + "::DakConfig")
84 if conffile: 84 ↛ 85line 84 didn't jump to line 85 because the condition on line 84 was never true
85 apt_pkg.read_config_file_isc(self.Cnf, conffile) # type: ignore[attr-defined]
87 # Read group-specific options
88 if "ByGroup" in self.Cnf: 88 ↛ 89line 88 didn't jump to line 89 because the condition on line 88 was never true
89 bygroup = self.Cnf.subtree("ByGroup") # type: ignore[attr-defined]
90 groups = {os.getgid()}
91 groups.update(os.getgroups())
93 for group in bygroup.list():
94 gid = grp.getgrnam(group).gr_gid
95 if gid in groups:
96 if bygroup.get(group):
97 apt_pkg.read_config_file_isc(self.Cnf, bygroup[group]) # type: ignore[attr-defined]
98 break
100 # Read user-specific options
101 byuser_config_path = self.Cnf.get(f"ByUser::{getpass.getuser()}", "")
102 if byuser_config_path: 102 ↛ 103line 102 didn't jump to line 103 because the condition on line 102 was never true
103 apt_pkg.read_config_file_isc(self.Cnf, byuser_config_path) # type: ignore[attr-defined]
105 if "Include" in self.Cnf: 105 ↛ 106line 105 didn't jump to line 106 because the condition on line 105 was never true
106 for filename in self.Cnf.value_list("Include"):
107 apt_pkg.read_config_file_isc(self.Cnf, filename) # type: ignore[attr-defined]
109 # Rebind some functions
110 # TODO: Clean this up
111 self.get = self.Cnf.get
112 self.subtree = self.Cnf.subtree # type: ignore[attr-defined]
113 self.value_list = self.Cnf.value_list
114 self.find = self.Cnf.find
115 self.find_b = self.Cnf.find_b
116 self.find_i = self.Cnf.find_i # type: ignore[attr-defined]
118 def __contains__(self, name: str) -> bool:
119 return name in self.Cnf
121 def __getitem__(self, name: str) -> str:
122 return self.Cnf[name]
124 def __setitem__(self, name: str, value: str) -> None:
125 self.Cnf[name] = value # type: ignore[index]
127 @staticmethod
128 def get_db_value(name, default=None, rettype=None):
129 from sqlalchemy import select
131 from daklib.dbconn import DBConfig, DBConn
133 q = select(DBConfig).where(DBConfig.name == name)
134 res = DBConn().session().execute(q).scalar_one_or_none()
135 if res is None: 135 ↛ 136line 135 didn't jump to line 136 because the condition on line 135 was never true
136 return default
138 if rettype: 138 ↛ 141line 138 didn't jump to line 141 because the condition on line 138 was always true
139 return rettype(res.value)
140 else:
141 return res.value
143 def _setup_routines(self):
144 """
145 This routine is the canonical list of which fields need to exist in
146 the config table. If your dak instance is to work, we suggest reading it
148 Of course, what the values do is another matter
149 """
150 for field in [
151 ("db_revision", None, int),
152 ("defaultsuitename", "unstable", str),
153 ("use_extfiles", None, int),
154 ]:
155 setattr(
156 self,
157 "get_%s" % field[0],
158 lambda s=None, x=field[0], y=field[1], z=field[2]: self.get_db_value(
159 x, y, z
160 ),
161 )
162 setattr(
163 Config,
164 "%s" % field[0],
165 property(fget=getattr(self, "get_%s" % field[0])),
166 )
168 def get_defaultsuite(self) -> "Suite | None":
169 from daklib.dbconn import get_suite
171 suitename = self.defaultsuitename # type: ignore[attr-defined]
172 if not suitename:
173 return None
174 else:
175 return get_suite(suitename)
177 defaultsuite = property(get_defaultsuite)