Tested processing by directory processor of templates with different package parameters and with merge parameters. The refactoring of the directory processor is started.

packages
Иванов Денис 4 years ago
parent 4b2dccabf9
commit dfe145725b

@ -1,7 +1,8 @@
# vim: fileencoding=utf-8
#
from pprint import pprint
from ..utils.package import PackageAtomParser, Package, PackageNotFound
from ..utils.package import PackageAtomParser, Package, PackageNotFound,\
PackageAtomName
from ..utils.files import join_paths, write_file, read_file_lines, FilesError,\
check_directory_link, read_link, Process,\
get_target_from_link
@ -330,9 +331,11 @@ class TemplateWrapper:
parameter_package = None
if self.target_type is not None:
print('looking for package for file: {}'.format(self.target_path))
try:
file_package = self.package_atom_parser.get_file_package(
self.target_path)
print('package is found: {}'.format(file_package))
except PackageNotFound:
file_package = None
self.target_without_package = True
@ -569,8 +572,8 @@ class TemplateWrapper:
def save_changes(self):
'''Метод для сохранения изменений внесенных в CONTENTS.'''
print('in save changes method')
if self.target_package:
print('SAVING TO CONTENTS: {}'.format(self.target_package))
self.target_package.remove_empty_directories()
self.target_package.write_contents_file()
@ -634,8 +637,6 @@ class TemplateExecutor:
target_package=None):
'''Метод для запуска выполнения шаблонов.'''
# Словарь с данными о результате работы исполнительного метода.
print('PARAMETERS: {}'.format(parameters))
print('PACKAGE: {}'.format(target_package))
self.executor_output = {'target_path': None,
'stdout': None,
'stderr': None}
@ -658,6 +659,7 @@ class TemplateExecutor:
# Удаляем оригинал, если это необходимо из-за наличия force или по
# другим причинам.
if template_object.remove_original:
print('REMOVE ORIGINAL')
if template_object.target_type == DIR:
self._remove_directory(template_object.target_path)
else:
@ -788,7 +790,6 @@ class TemplateExecutor:
output_path = template_object.output_path
template_format = template_object.format_class
print('JOIN FILE')
# Задаемся значениями chmod и chown в зависимости от наличия или
# отсутствия файла, принадлежности его пакету и наличия значений
@ -810,16 +811,13 @@ class TemplateExecutor:
chown = self.file_default_parameters.get('chown', False)
if template_format.EXECUTABLE or template_object.md5_matching:
print('MD5 MATCHES')
# Действия при совпадении md5 из CONTENTS и md5 целевого файла.
# А также если шаблон просто исполнительный.
output_paths = [output_path]
# Если целевой файл защищен, а шаблон не userspace.
print('PROTECTED: {}'.format(template_object._protected_set))
if template_object.protected and not template_object.is_userspace:
# Тогда также обновляем архив.
print('SAVE ARCHIVE: {}'.format(template_object.archive_path))
output_paths.append(template_object.archive_path)
if template_object.target_type is not None and not replace:
@ -827,17 +825,14 @@ class TemplateExecutor:
# текст целевого файла.
if (not input_path.startswith(self.cl_config_archive_path) or
os.path.exists(input_path)):
print('INPUT FROM FILE')
# Если входной файл просто не из архива, или из архива и
# при этом существует -- используем его
with open(input_path, 'r') as input_file:
input_text = input_file.read()
else:
print('INPUT IS EMPTY ARCHIVE FILE')
# В противном случае используем пустой файл. (!)
input_text = ''
else:
print('INPUT IS NEW EMPTY FILE')
input_text = ''
parsed_template = template_format(template_object.template_text,
@ -1584,12 +1579,17 @@ class DirectoryProcessor:
appends_set=self.template_executor.available_appends)
if package:
try:
self.for_package = self.template_engine.parameters_processor.\
check_package_parameter(package)
except ConditionFailed as error:
self.output.set_error(str(error))
return
if isinstance(package, PackageAtomName):
self.for_package = package
elif isinstance(package, str):
try:
self.for_package = self.template_engine.\
parameters_processor.check_package_parameter(package)
except ConditionFailed as error:
self.output.set_error(str(error))
return
else:
self.for_package = False
else:
self.for_package = False
@ -1642,8 +1642,6 @@ class DirectoryProcessor:
for node in entries:
self.directory_tree = {}
print('Empty parameters container: {}'.format(
ParametersContainer()))
self.walk_directory_tree(node.path,
self.cl_chroot_path,
ParametersContainer(),
@ -1690,7 +1688,7 @@ class DirectoryProcessor:
def get_directories_queue(self, path):
'''Уже не актуальный метод для построение очередей из путей к
шаблонам.'''
шаблонам. Хотя возможно еще пригодится.'''
directories_queue = []
for base_dir in self.template_paths:
@ -1715,6 +1713,9 @@ class DirectoryProcessor:
def walk_directory_tree(self, current_directory_path, target_path,
directory_parameters, directory_tree={}):
'''Метод для рекурсивного обхода директорий с шаблонами, а также, при
необходимости, заполнения деревьев директорий шаблонов, с помощью
которых далее выполняются шаблоны пакетов из merge.'''
print('current_directory: {}'.format(current_directory_path))
print('fill trees = {}'.format(self.fill_trees))
template_files = []
@ -1731,12 +1732,8 @@ class DirectoryProcessor:
self.template_engine.change_directory(current_directory_path)
for node in entries:
print('check: {}'.format(node.name))
if self.check_file_name(node.name):
print('ignore it')
continue
else:
print('all is good')
if node.is_symlink():
self.output.set_warning('symlink: {} is ignored'.
@ -1757,13 +1754,13 @@ class DirectoryProcessor:
template_type=DIR,
parameters=directory_parameters)
except ConditionFailed as error:
self.output.set_warning('{}.\nTemplate: {}'.
self.output.set_warning('{}. Template: {}'.
format(str(error),
current_directory_path))
directory_tree = {}
return
except Exception as error:
self.output.set_error('Template error: {0}\nTemplate: {1}'.
self.output.set_error('Template error: {0} Template: {1}'.
format(str(error),
current_directory_path))
directory_tree = {}
@ -1781,77 +1778,102 @@ class DirectoryProcessor:
current_target_path = join_paths(self.cl_chroot_path,
directory_parameters.path)
if self.fill_trees and self.check_package_and_action(
directory_parameters,
current_directory_path,
directory_tree=directory_tree):
if not directory_parameters.append == 'skip':
current_target_path = os.path.join(current_target_path,
directory_name)
elif not self.fill_trees and self.check_package_and_action(
directory_parameters,
current_directory_path):
# В зависимости от того, нужно ли нам заполнять дерево директорий,
# отправляем в метод для проверки параметров package и action
# или текущее дерево, или пустое значение.
if self.fill_trees:
tree_for_checker = directory_tree
else:
tree_for_checker = None
if self.check_package_and_action(directory_parameters,
current_directory_path,
directory_tree=tree_for_checker):
# Если проверка параметров package и action пройдена и
# и параметр append не равен skip -- обновляем целевой путь
# именем текущей директории.
if not directory_parameters.append == 'skip':
current_target_path = os.path.join(current_target_path,
directory_name)
else:
# Обновляем дерево директорий для данного пакета.
# Если проверка не пройдена, включено заполнение дерева, а
# метод для проверки заменил для текущей директории значение {}
# на None (являющейся для этого дерева заглушкой), то,
# используя нынешнее состояние дерева директорий, обновляем
# дерево пакета текущего шаблона директории.
if (self.fill_trees and
directory_tree[directory_name] is None):
package_name = directory_parameters.package
if package_name in self.packages_file_trees:
# Если для данного пакета уже есть дерево --
# накладываем на него текущее.
self.packages_file_trees[package_name].update_tree(
copy.deepcopy(self.directory_tree)
)
else:
# Если для данного пакета еще нет дерева --
# копируем для него текущее.
directory_tree = DirectoryTree(self.base_directory)
directory_tree.update_tree(
copy.deepcopy(self.directory_tree))
self.packages_file_trees[package_name] = directory_tree
# Перед выходом из директории очищаем текущий уровень
# дерева.
directory_tree = {}
return
# Если есть параметр merge -- сохраняем присутствующие в нем пакеты
# для последующей обработки.
if self.for_package and directory_parameters.merge:
self.packages_to_merge.extend(directory_parameters.merge)
else:
# Если .calculate_directory отсутствует -- создаем директорию
# используя унаследованные параметры и имя самой директории.
if self.fill_trees:
tree_for_checker = directory_tree
else:
tree_for_checker = None
if not self.check_package_and_action(
directory_parameters,
current_directory_path,
directory_tree=directory_tree):
directory_parameters,
current_directory_path,
directory_tree=tree_for_checker):
# Обновляем дерево директорий для данного пакета.
if (directory_tree[directory_name] is None and
self.fill_trees):
package_name = directory_parameters.package
if package_name in self.packages_file_trees:
# Если для данного пакета уже есть дерево --
# накладываем на него текущее.
self.packages_file_trees[package_name].update_tree(
copy.deepcopy(self.directory_tree)
copy.deepcopy(self.directory_tree)
)
else:
# Если для данного пакета еще нет дерева --
# копируем для него текущее.
directory_tree = DirectoryTree(self.base_directory)
directory_tree.update_tree(
copy.deepcopy(self.directory_tree))
self.packages_file_trees[package_name] = directory_tree
# Перед выходом из директории очищаем текущий уровень
# дерева.
directory_tree = {}
return
# Для того чтобы директория была создана просто добавляем параметр
# append = join.
directory_parameters.set_parameter({'append': 'join'})
current_target_path = os.path.join(current_target_path,
directory_name)
# Выполняем действие с директорией.
self.output.set_success('Processing directory: {}'.
format(current_directory_path))
print('target_path: {}'.format(current_target_path))
# * * * МОЖНО ВЫДЕЛИТЬ В ОТДЕЛЬНЫЙ МЕТОД * * *
# Выполняем наложение шаблона.
try:
output = self.template_executor.execute_template(
current_target_path,
directory_parameters,
DIR)
current_target_path,
directory_parameters,
DIR)
# Если во время выполнения шаблона был изменен целевой путь,
# например, из-за ссылки на директорию в source -- обновляем
# целевой путь.
@ -1860,27 +1882,34 @@ class DirectoryProcessor:
# Если есть вывод от параметра run -- выводим как info.
if output['stdout'] is not None:
self.output.set_info("stdout from template: {}\n{}".format(
self.output.set_info("stdout from template: {}:\n{}\n".format(
current_directory_path,
output['stdout']
))
# Если есть ошибки от параметра run -- выводим их как error
# Если есть ошибки от параметра run -- выводим их как error.
if output['stderr'] is not None:
self.output.set_error("stderr from template: {}\n{}".
self.output.set_error("stderr from template: {}:\n{}\n".
format(current_directory_path,
output['stderr']))
# Если run выполнен с ошибками -- пропускаем директорию.
return
except TemplateTypeConflict as error:
self.output.set_error('Type conflict: {}\nTemplate: {}'.
except TemplateExecutorError as error:
self.output.set_error('Template execution error: {} Template: {}'.
format(str(error),
current_directory_path))
return
# Если режим заполнения очередей выключен и дерево, которое обходим в
# данный момент еще не пусто -- используем имеющееся дерево для
# получения списков обрабатываемых файлов и директорий.
self.output.set_success('Processing directory: {}'.
format(current_directory_path))
# * * * МОЖНО ВЫДЕЛИТЬ В ОТДЕЛЬНЫЙ МЕТОД * * *
# Далее обрабатываем файлы шаблонов хранящихся в директории.
# Список файлов и директорий для дальнейшего обхода получаем из
# списоков полученных при начальном сканировании директории или,
# если в данный момент обходим дерево директорий, что можно определить
# по выключенному флагу fill_trees и по наличию дерева в переменной
# directory_tree, берем файлы и каталоги для обхода из дерева.
if not self.fill_trees and directory_tree:
tree_files = []
tree_directories = []
@ -1896,7 +1925,7 @@ class DirectoryProcessor:
template_files = tree_files
template_directories = tree_directories
# обрабатываем файлы шаблонов хранящихся в директории.
# Обрабатываем файлы шаблонов.
for template_name in template_files:
# Удаляем все параметры, которые не наследуются и используем
# полученный контейнер для сбора параметров файлов шаблонов.
@ -1910,43 +1939,44 @@ class DirectoryProcessor:
template_name, FILE,
parameters=directory_parameters)
except ConditionFailed as error:
self.output.set_warning('{0}.\nTemplate: {1}'.
self.output.set_warning('{0}. Template: {1}'.
format(str(error),
current_directory_path))
template_path))
continue
except Exception as error:
self.output.set_error('Template error: {0} \nTemplate: {1}'.
self.output.set_error('Template error: {0} Template: {1}'.
format(str(error),
current_directory_path))
template_path))
continue
# directory_parameters.print_parameters_for_debug()
directory_parameters.print_parameters_for_debug()
template_text = self.template_engine.template_text
if self.fill_trees and not self.check_package_and_action(
# Если находимся на стадии заполнения дерева директорий --
# проверяем параметры package и action с заполнением дерева.
if self.fill_trees:
tree_for_checker = directory_tree[directory_name]
else:
tree_for_checker = None
# Проверяем параметры action и package.
if not self.check_package_and_action(
directory_parameters,
template_path,
directory_tree=directory_tree[directory_name]):
# Если находимся на стадии заполнения дерева директорий --
# проверяем параметры package и action с заполнением дерева.
directory_tree=tree_for_checker):
continue
elif not self.fill_trees and not self.check_package_and_action(
directory_parameters,
template_path):
# В противном случае проверяем без заполнения.
continue
# Если есть параметр merge добавляем в список
# Если есть параметр merge добавляем в список пакетов для
# последующей обработки.
if self.for_package and directory_parameters.merge:
self.packages_to_merge.extend(directory_parameters.merge)
# Если для шаблона задано имя -- меняем его
# Если для шаблона задано имя -- меняем его.
if directory_parameters.name:
template_name = directory_parameters.name
# Если для шаблона задан путь -- меняем его
# Если для шаблона задан путь -- меняем его.
if directory_parameters.path:
target_file_path = join_paths(self.cl_chroot_path,
directory_parameters.path,
@ -1956,11 +1986,18 @@ class DirectoryProcessor:
template_name)
# Выполняем действия, указанные в шаблоне.
output = self.template_executor.execute_template(
target_file_path,
directory_parameters,
FILE,
template_text=template_text)
try:
output = self.template_executor.execute_template(
target_file_path,
directory_parameters,
FILE,
template_text=template_text)
except TemplateExecutorError as error:
self.output.set_error(
'Template execution error: {0} Template: {1}'.
format(str(error),
template_path))
continue
# Если во время выполнения шаблона был изменен целевой путь,
# например, из-за ссылки на директорию в source -- обновляем
@ -1986,22 +2023,25 @@ class DirectoryProcessor:
self.output.set_success('Processed template: {}...'.
format(template_path))
# Обновляем дерево директорий для данного пакета, если происходит
# его заполнение.
if (self.fill_trees and directory_tree[directory_name]):
package_name = directory_parameters.package
# Обновляем дерево директорий для данного пакета, если происходит
# его заполнение.
self.output.set_info("Let's update tree using current: {}".
format(directory_tree))
if (self.fill_trees and directory_tree[directory_name]):
package_name = directory_parameters.package
if package_name in self.packages_file_trees:
# Если для данного пакета дерево уже есть -- обновляем его.
self.packages_file_trees[package_name].update_tree(
if package_name in self.packages_file_trees:
# Если для данного пакета дерево уже есть -- обновляем его.
self.packages_file_trees[package_name].update_tree(
copy.deepcopy(self.directory_tree))
else:
# Если нет создаем новое.
directory_tree = DirectoryTree(self.base_directory)
directory_tree.update_tree(copy.deepcopy(self.directory_tree))
self.packages_file_trees[package_name] = directory_tree
else:
# Если нет создаем новое.
directory_tree = DirectoryTree(self.base_directory)
directory_tree.update_tree(copy.deepcopy(
self.directory_tree))
self.packages_file_trees[package_name] = directory_tree
directory_tree[directory_name] = {}
directory_tree[directory_name] = {}
# проходимся далее по директориям.
for directory in template_directories:
@ -2042,7 +2082,7 @@ class DirectoryProcessor:
elif parameters.action != self.action:
self.output.set_warning(
("Action parameter value '{0}' does not match its"
" current value '{1}'.\nTemplate: {2}").format(
" current value '{1}'. Template: {2}").format(
parameters.action,
self.action,
template_path))
@ -2051,21 +2091,25 @@ class DirectoryProcessor:
if self.for_package:
if not parameters.package:
self.output.set_warning(
"'package' parameter is not defined.\nTemplate: {}".
"'package' parameter is not defined. Template: {}".
format(template_path))
# считаем наличие параметра package необязательным на данном
# этапе.
# пока что считаем наличие параметра package необязательным
# return False
elif parameters.package != self.for_package:
if directory_tree is not None:
template_name = os.path.basename(template_path)
directory_tree[template_name] = None
self.output.set_warning('DirectoryTree: {}'.
format(directory_tree))
self.output.set_warning(
("'package' parameter value '{0}' does not "
"match its current target package.\nTemplate: {1}").
format(parameters.package, template_path)
"match its current target package '{1}'. "
"Template: {2}").
format(parameters.package.atom,
self.for_package.atom,
template_path)
)
return False

@ -3,7 +3,6 @@ import re
from . import files
from .tools import Cachable, GenericFS, unique
from time import sleep
from pprint import pprint
def get_uuid_dict(reverse=False, devices=()):

@ -9,10 +9,11 @@ import os
import sys
import contextlib
def listDirectory(dn, fullpath=False):
if path.exists(dn):
if fullpath:
return [path.join(dn,x) for x in os.listdir(dn)]
return [path.join(dn, x) for x in os.listdir(dn)]
else:
return os.listdir(dn)
return []
@ -561,6 +562,7 @@ def get_run_commands(not_chroot=False, chroot=None, uid=None, with_pid=False):
output.append(cmdline)
return output
@contextlib.contextmanager
def stderr_devnull():
oldstderr = sys.stderr

@ -2,6 +2,7 @@ import os
import re
import calculate.utils.fs as fs
class ProfileWalker:
"""
Объект обходящий все директории профиля через parent файлы
@ -9,11 +10,12 @@ class ProfileWalker:
def __init__(self, filename, repositories):
self.repositories = repositories
self.filename = filename
self.reReppath = re.compile("^(%s)+:"%"|".join(self.repositories.keys()))
self.reReppath = re.compile("^({0})+:".format(
"|".join(self.repositories.keys())))
def interpolate(self, path):
def subfunc(m):
return "%s/profiles/" % self.repositories.get(m.group(1))
return "{0}/profiles/".format(self.repositories.get(m.group(1)))
return self.reReppath.sub(subfunc, path)
def find(self, directory):

@ -2,7 +2,8 @@ import pytest
import shutil
import hashlib
import os
from calculate.templates.template_processor import DirectoryProcessor
from calculate.templates.template_processor import DirectoryProcessor,\
TemplateWrapper
from calculate.utils.package import PackageAtomName, Version, Package
from calculate.utils.files import join_paths
from calculate.utils.io_module import IOModule
@ -35,7 +36,6 @@ new_package_name = PackageAtomName(
'var/db/pkg/test-category/new-package-0.1.1'),
'version': Version('1.0')})
# Вместо модуля переменных.
group = Variables({'bool': True,
'list': [1, 2, 3]})
@ -91,6 +91,8 @@ def show_tree(dir_path, indent=0):
@pytest.mark.directory_processor
class TestDirectoryProcessor:
def test_create_testfiles(self):
TemplateWrapper._protected_is_set = False
shutil.copytree(os.path.join(CHROOT_PATH, 'etc.backup'),
os.path.join(CHROOT_PATH, 'etc'),
symlinks=True)
@ -105,7 +107,7 @@ class TestDirectoryProcessor:
datavars_module=datavars)
directory_processor.process_template_directories()
def if_templates_consist_only_one_directory_with_a_calculate_directory_file_and_a_single_template_file__the_directory_processor_creates_new_file_and_adds_one_in_the_CONTENTS_file(self):
def test_if_templates_consist_only_one_directory_with_a_calculate_directory_file_and_a_single_template_file__the_directory_processor_creates_new_file_and_adds_one_in_the_CONTENTS_file(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
'templates_1')
directory_processor = DirectoryProcessor('install',
@ -123,7 +125,7 @@ class TestDirectoryProcessor:
output_file_text = output_file.read()
assert output_file_text == output_text
def if_templates_consist_only_one_directory_with_a_calculate_directory_file_and_a_single_other_directory_with_same_a_file__the_directory_processor_creates_new_directory(self):
def test_if_templates_consist_only_one_directory_with_a_calculate_directory_file_and_a_single_other_directory_with_same_a_file__the_directory_processor_creates_new_directory(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
'templates_2')
directory_processor = DirectoryProcessor('install',
@ -132,7 +134,7 @@ class TestDirectoryProcessor:
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_1'))
def if_templates_consist_only_one_directory_with_a_calculate_directory_file_and_a_single_other_directory_without_calculate_directory_file__the_directory_processor_creates_new_directory(self):
def test_if_templates_consist_only_one_directory_with_a_calculate_directory_file_and_a_single_other_directory_without_calculate_directory_file__the_directory_processor_creates_new_directory(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
'templates_4')
directory_processor = DirectoryProcessor('install',
@ -141,7 +143,7 @@ class TestDirectoryProcessor:
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_3'))
def if_templates_consist_only_one_directory_with_a_calculate_directory_file_with_a_single_directory_with_a_single_template_file__the_directory_processor_creates_new_directory_and_file_and_adds_one_in_the_CONTENTS_file(self):
def test_if_templates_consist_only_one_directory_with_a_calculate_directory_file_with_a_single_directory_with_a_single_template_file__the_directory_processor_creates_new_directory_and_file_and_adds_one_in_the_CONTENTS_file(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
'templates_3')
directory_processor = DirectoryProcessor('install',
@ -162,7 +164,7 @@ class TestDirectoryProcessor:
output_file_text = output_file.read()
assert output_file_text == output_text
def if_templates_consist_only_one_directory_with_a_calculate_directory_file_and_with_two_directories_with_a_template_files__the_directory_processor_creates_all_new_directories_and_files_and_adds_them_in_the_CONTENTS_file(self):
def test_if_templates_consist_only_one_directory_with_a_calculate_directory_file_and_with_two_directories_with_a_template_files__the_directory_processor_creates_all_new_directories_and_files_and_adds_them_in_the_CONTENTS_file(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
'templates_5')
directory_processor = DirectoryProcessor('install',
@ -197,7 +199,7 @@ class TestDirectoryProcessor:
output_file_text = output_file.read()
assert output_file_text == output_text
def if_templates_consist_only_one_directory_with_a_calculate_directory_file_and_a_single_template_file_and_there_is_a_file_without_user_changes_on_its_target_path__the_directory_processor_joins_a_template_file_with_a_target_file_and_updates_the_CONTENTS_file(self):
def test_if_templates_consist_only_one_directory_with_a_calculate_directory_file_and_a_single_template_file_and_there_is_a_file_without_user_changes_on_its_target_path__the_directory_processor_joins_a_template_file_with_a_target_file_and_updates_the_CONTENTS_file(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
'templates_6')
directory_processor = DirectoryProcessor('install',
@ -218,7 +220,7 @@ class TestDirectoryProcessor:
output_file_text = output_file.read()
assert output_file_text == output_text
def if_templates_consist_only_one_directory_with_a_calculate_directory_file_with_a_single_directory_with_a_single_template_file_and_there_is_a_file_without_user_changes_on_its_target_path__the_directory_processor_joins_a_template_file_with_a_target_file_and_updates_the_CONTENTS_file(self):
def test_if_templates_consist_only_one_directory_with_a_calculate_directory_file_with_a_single_directory_with_a_single_template_file_and_there_is_a_file_without_user_changes_on_its_target_path__the_directory_processor_joins_a_template_file_with_a_target_file_and_updates_the_CONTENTS_file(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
'templates_7')
directory_processor = DirectoryProcessor('install',
@ -242,7 +244,7 @@ class TestDirectoryProcessor:
hashlib.md5(
output_text.encode()).hexdigest())
def if_templates_consist_only_one_directory_with_a_calculate_directory_file_and_a_single_template_file_and_there_is_a_file_with_user_changes_on_its_target_path__the_directory_processor_joins_a_template_file_with_a_target_file_and_updates_the_CONTENTS_file(self):
def test_if_templates_consist_only_one_directory_with_a_calculate_directory_file_and_a_single_template_file_and_there_is_a_file_with_user_changes_on_its_target_path__the_directory_processor_joins_a_template_file_with_a_target_file_and_updates_the_CONTENTS_file(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
'templates_8')
directory_processor = DirectoryProcessor('install',
@ -266,7 +268,7 @@ class TestDirectoryProcessor:
hashlib.md5(
output_text.encode()).hexdigest())
def if_templates_consist_only_one_directory_with_a_calculate_directory_file_one_template_directory_and_a_single_template_file_with_a_target_path_to_a_file_removed_by_user_in_the_last_one__the_directory_processor_creates_a_new_empty_cfg_file__joins_template_with_it_and_updates_the_CONTENTS_file(self):
def test_if_templates_consist_only_one_directory_with_a_calculate_directory_file_one_template_directory_and_a_single_template_file_with_a_target_path_to_a_file_removed_by_user_in_the_last_one__the_directory_processor_creates_a_new_empty_cfg_file__joins_template_with_it_and_updates_the_CONTENTS_file(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
'templates_9')
directory_processor = DirectoryProcessor('install',
@ -290,7 +292,7 @@ class TestDirectoryProcessor:
hashlib.md5(
output_text.encode()).hexdigest())
def if_templates_are_hierarchy_of_a_multiple_template_files_with_a_removed_or_changed_by_user_targets_and_there_is_the_autoupdate_parameter_in_the_root_calculate_directory_template__the_directory_processor_uses_autoupdate_parameter_for_all_templates_and_joins_all_templates_as_if_target_files_have_no_user_changes(self):
def test_if_templates_are_hierarchy_of_a_multiple_template_files_with_a_removed_or_changed_by_user_targets_and_there_is_the_autoupdate_parameter_in_the_root_calculate_directory_template__the_directory_processor_uses_autoupdate_parameter_for_all_templates_and_joins_all_templates_as_if_target_files_have_no_user_changes(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
'templates_10')
directory_processor = DirectoryProcessor('install',
@ -342,7 +344,7 @@ class TestDirectoryProcessor:
hashlib.md5(
output_text_2.encode()).hexdigest())
def if_the_template_directory_have_no_the_action_parameter_value_and_append_parameter_is_not_skip__the_directory_processor_skips_this_template_branch_and_sets_warning(self):
def test_if_the_template_directory_have_no_the_action_parameter_value_and_append_parameter_is_not_skip__the_directory_processor_skips_this_template_branch_and_sets_warning(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
'templates_11')
io_module = IOModule()
@ -357,12 +359,12 @@ class TestDirectoryProcessor:
assert not os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_8'))
assert io_module.messages[-1] == ('warning', warning_message)
def if_the_template_has_two_root_directories_with_different_action_values_and_directory_processor_intialized_for_the_one_of_this_actions__the_directory_processor_skips_one_this_template_s_roots_and_processed_a_template_s_root_with_the_same_action_parameter_value(self):
def test_if_the_template_has_two_root_directories_with_different_action_values_and_directory_processor_intialized_for_the_one_of_this_actions__the_directory_processor_skips_one_this_template_s_roots_and_processed_a_template_s_root_with_the_same_action_parameter_value(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
'templates_12')
io_module = IOModule()
warning_message = ("Action parameter value '{0}' does not match its"
" current value '{1}'.\nTemplate: {2}").format(
" current value '{1}'. Template: {2}").format(
'update',
'install',
join_paths(
@ -376,6 +378,239 @@ class TestDirectoryProcessor:
assert not os.path.exists(join_paths(CHROOT_PATH, '/etc/file_7'))
assert io_module.messages[-1] == ('warning', warning_message)
def test_if_some_template_directories_have_no_the_action_parameter_value_but_the_append_parameter_s_value_is_skip__the_directory_processor_does_not_stop_the_directories_processing_and_sends_no_warnings(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
'templates_13')
directory_processor = DirectoryProcessor('install',
datavars_module=datavars)
directory_processor.process_template_directories()
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/file_8'))
def test_if_template_s_directory_contains_two_directories_with_single_template_files_that_belongs_to_a_different_packages_and_target_files_does_not_exist__the_directory_processor_creates_two_files_and_adds_them_to_a_different_packages(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
'templates_14')
directory_processor = DirectoryProcessor('install',
datavars_module=datavars)
directory_processor.process_template_directories()
test_package = Package(test_package_name, chroot_path=CHROOT_PATH)
other_package = Package(other_package_name, chroot_path=CHROOT_PATH)
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_9'))
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_9/file_0'))
assert '/etc/dir_9/file_0' in other_package
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_10'))
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_10/file_0'))
assert '/etc/dir_10/file_0' in test_package
def test_if_template_s_directory_contains_one_directory_with_a_template_file_without_a_package_value_target_file_does_not_exist_and_template_executor_is_not_able_to_detect_package_using_target_path__the_directory_processor_skips_this_template_and_sets_error_in_the_output(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
'templates_15')
error_message = ("Template execution error: collision: 'package' "
"parameter is not defined for template with 'append' "
"parameter. Template: /home/divanov/Home/development/"
"calculate-lib/tests/templates/testfiles/"
"test_dir_processor_root/templates_15/root/dir_11")
io_module = IOModule()
directory_processor = DirectoryProcessor('install',
datavars_module=datavars,
output_module=io_module)
directory_processor.process_template_directories()
assert not os.path.exists(join_paths(CHROOT_PATH,
'/etc/dir_11'))
assert not os.path.exists(join_paths(CHROOT_PATH,
'/etc/dir_11/file_0'))
assert io_module.messages[-1] == ('error', error_message)
def test_if_template_s_directory_contains_one_directory_with_a_template_file_without_a_package_value_target_file_exists_and_template_executor_is_able_to_detect_package_using_target_path__the_directory_processor_joins_template_to_a_target_file_and_updates_the_CONTENTS_file(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
'templates_16')
directory_processor = DirectoryProcessor('install',
datavars_module=datavars)
directory_processor.process_template_directories()
new_package = Package(new_package_name, chroot_path=CHROOT_PATH)
assert os.path.exists(join_paths(CHROOT_PATH,
'/etc/dir_12'))
assert os.path.exists(join_paths(CHROOT_PATH,
'/etc/dir_12/file_0'))
assert '/etc/dir_12/file_0' in new_package
def test_if_template_s_directory_contains_two_directories_with_single_template_files_that_belongs_to_a_different_packages_and_target_files_does_not_exist_and_directory_processor_is_used_for_a_package__the_directory_processor_creates_one_file_using_template_with_actual_package_parameter_and_adds_it_to_a_package_and_add_to_the_packages_file_trees_a_directory_with_an_other_template(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
'templates_17')
directory_processor = DirectoryProcessor('install',
datavars_module=datavars,
package=other_package_name)
directory_processor.process_template_directories()
test_package = Package(test_package_name, chroot_path=CHROOT_PATH)
other_package = Package(other_package_name, chroot_path=CHROOT_PATH)
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_13'))
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_13/file_0'))
assert '/etc/dir_13/file_0' in other_package
assert not os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_14'))
assert not os.path.exists(join_paths(CHROOT_PATH,
'/etc/dir_14/file_0'))
assert '/etc/dir_14/file_0' not in test_package
def test_if_template_s_directory_contains_two_directories_with_single_template_files_that_belongs_to_a_different_packages_and_target_files_does_not_exist_and_one_of_a_template_file_has_the_merge_parameter_with_other_package_and_directory_processor_is_used_for_a_package__the_directory_processor_creates_one_file_using_template_with_actual_package_parameter_and_then_uses_the_packages_file_tree_to_merge_an_other_package(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
'templates_18')
directory_processor = DirectoryProcessor('install',
datavars_module=datavars,
package=other_package_name)
directory_processor.process_template_directories()
test_package = Package(test_package_name, chroot_path=CHROOT_PATH)
other_package = Package(other_package_name, chroot_path=CHROOT_PATH)
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_15'))
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_15/file_0'))
assert '/etc/dir_15/file_0' in other_package
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_16'))
assert os.path.exists(join_paths(CHROOT_PATH,
'/etc/dir_16/file_0'))
assert '/etc/dir_16/file_0' in test_package
def test_if_template_s_directory_contains_a_template_directory_which_target_is_a_link_to_an_other_directory_and_force_parameter_is_not_set__the_directory_processor_changes_a_template_target_path_to_a_link_source_path_and_joins_all_templates_from_the_template_directory(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
'templates_19')
directory_processor = DirectoryProcessor('install',
datavars_module=datavars)
directory_processor.process_template_directories()
test_package = Package(test_package_name, chroot_path=CHROOT_PATH)
# Для разнообразия один из шаблонов удаляет файл, а не создает.
assert not os.path.exists(join_paths(CHROOT_PATH,
'/etc/dir_18/file_0'))
assert '/etc/dir_18/file_0' not in test_package
assert os.path.exists(join_paths(CHROOT_PATH,
'/etc/dir_18/dir_0'))
assert os.path.exists(join_paths(CHROOT_PATH,
'/etc/dir_18/dir_0/file_0'))
assert '/etc/dir_18/dir_0/file_0' in test_package
def test_if_template_s_directory_contains_a_template_directory_which_target_is_a_link_to_an_other_directory_and_force_parameter_is_set__the_directory_processor_removes_link_on_a_target_path_and_joins_all_templates_from_a_template_directory(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
'templates_20')
directory_processor = DirectoryProcessor('install',
datavars_module=datavars)
directory_processor.process_template_directories()
test_package = Package(test_package_name, chroot_path=CHROOT_PATH)
# Для разнообразия один из шаблонов удаляет файл, а не создает.
# Но в данном случае он ничего не сделает.
assert os.path.exists(join_paths(CHROOT_PATH,
'/etc/dir_20/file_0'))
assert '/etc/dir_20/file_0' not in test_package
assert not os.path.islink(join_paths(CHROOT_PATH, '/etc/dir_19'))
assert os.path.isdir(join_paths(CHROOT_PATH, '/etc/dir_19'))
assert os.path.exists(join_paths(CHROOT_PATH,
'/etc/dir_19/dir_0'))
assert os.path.exists(join_paths(CHROOT_PATH,
'/etc/dir_19/dir_0/file_0'))
assert '/etc/dir_19/dir_0/file_0' in test_package
def test_if_template_s_directory_contains_some_directories_with_single_template_files_that_belongs_to_a_different_packages_and_target_files_does_not_exist_and_one_of_a_template_file_has_the_merge_parameter_with_other_packages_and_directory_processor_is_used_for_a_package__the_directory_processor_creates_one_file_using_template_with_actual_package_parameter_and_then_uses_the_packages_file_trees_to_merge_other_packages(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
'templates_21')
directory_processor = DirectoryProcessor('install',
datavars_module=datavars,
package=other_package_name)
directory_processor.process_template_directories()
test_package = Package(test_package_name, chroot_path=CHROOT_PATH)
other_package = Package(other_package_name, chroot_path=CHROOT_PATH)
new_package = Package(new_package_name, chroot_path=CHROOT_PATH)
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_21'))
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_21/file_0'))
assert '/etc/dir_21/file_0' in other_package
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_22'))
assert os.path.exists(join_paths(CHROOT_PATH,
'/etc/dir_22/file_0'))
assert '/etc/dir_22/file_0' in test_package
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_23'))
assert os.path.exists(join_paths(CHROOT_PATH,
'/etc/dir_23/file_0'))
assert '/etc/dir_23/file_0' in new_package
def test_if_template_s_directory_contains_some_directories_with_single_template_files_that_belong_to_a_different_packages_and_target_files_does_not_exist_and_some_of_a_template_files_have_the_merge_parameters_with_other_packages_and_directory_processor_is_used_for_a_package__the_directory_processor_creates_one_file_using_template_with_actual_package_parameter_and_then_uses_the_packages_file_trees_to_merge_other_packages(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
'templates_22')
directory_processor = DirectoryProcessor(
'install',
datavars_module=datavars,
package='test-category/other-package')
directory_processor.process_template_directories()
test_package = Package(test_package_name, chroot_path=CHROOT_PATH)
other_package = Package(other_package_name, chroot_path=CHROOT_PATH)
new_package = Package(new_package_name, chroot_path=CHROOT_PATH)
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_24'))
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_24/file_0'))
assert '/etc/dir_24/file_0' in other_package
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_25'))
assert os.path.exists(join_paths(CHROOT_PATH,
'/etc/dir_25/file_0'))
assert '/etc/dir_25/file_0' in test_package
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_26'))
assert os.path.exists(join_paths(CHROOT_PATH,
'/etc/dir_26/file_0'))
assert '/etc/dir_26/file_0' in new_package
print('DIRECTORY TREE:')
for key in directory_processor.packages_file_trees.keys():
print("{} -> {}".format(
key,
directory_processor.packages_file_trees[key]))
def test_if_template_s_directory_contains_some_directories_with_single_template_files_and_file_that_belong_to_a_different_packages_and_target_files_does_not_exist_and_some_of_a_template_files_have_the_merge_parameters_with_other_packages_and_directory_processor_is_used_for_a_package__the_directory_processor_creates_one_file_using_template_with_actual_package_parameter_and_then_uses_the_packages_file_trees_to_merge_other_packages(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
'templates_23')
directory_processor = DirectoryProcessor(
'install',
datavars_module=datavars,
package='test-category/other-package')
directory_processor.process_template_directories()
test_package = Package(test_package_name, chroot_path=CHROOT_PATH)
other_package = Package(other_package_name, chroot_path=CHROOT_PATH)
print('DIRECTORY TREE:')
for key in directory_processor.packages_file_trees.keys():
print("{} -> {}".format(
key,
directory_processor.packages_file_trees[key]))
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_27'))
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_27/file_0'))
assert '/etc/dir_27/file_0' in test_package
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_28'))
assert os.path.exists(join_paths(CHROOT_PATH,
'/etc/dir_28/file_0'))
assert '/etc/dir_28/file_0' in other_package
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/file_9'))
assert '/etc/file_9' in test_package
def test_view_tree(self):
list_path = join_paths(CHROOT_PATH, '/etc')
show_tree(list_path)

@ -49,6 +49,8 @@ other_package_name = PackageAtomName(
@pytest.mark.template_executor
class TestTemplateExecutor:
def test_function_to_copy_testfiles(self):
TemplateWrapper._protected_is_set = False
shutil.copytree(os.path.join(CHROOT_PATH, 'etc.backup'),
os.path.join(CHROOT_PATH, 'etc'),
symlinks=True)

@ -1,2 +1,2 @@
{% calculate append = 'skip', package = 'test-category/test-package' %}
{% calculate path = '/etc', action = 'install' %}
{% calculate action = 'install' %}

@ -1,2 +1,2 @@
{% calculate append = 'skip', package = 'test-category/test-package' %}
{% calculate path = '/etc', action = 'update' %}
{% calculate action = 'update' %}

@ -0,0 +1 @@
{% calculate append = 'skip', package = 'test-category/test-package' %}

@ -0,0 +1,5 @@
{% calculate append = 'join', format = 'bind', path = '/etc', action = 'install' -%}
options {
parameter-1 {{ variables.variable_1 }};
parameter-2 {{ variables.variable_2 }};
};

@ -0,0 +1 @@
{% calculate append = 'skip', action = 'install', path = '/etc' %}

@ -0,0 +1 @@
{% calculate append = 'join', package = 'test-category/test-package' %}

@ -0,0 +1,5 @@
{% calculate append = 'join', format = 'bind' -%}
options {
parameter-1 {{ variables.variable_1 }};
parameter-2 {{ variables.variable_2 }};
};

@ -0,0 +1 @@
{% calculate append = 'join', package = 'test-category/other-package' %}

@ -0,0 +1,5 @@
{% calculate append = 'join', format = 'bind' -%}
options {
parameter-1 {{ variables.variable_1 }};
parameter-2 {{ variables.variable_2 }};
};

@ -0,0 +1 @@
{% calculate append = 'skip', action = 'install', path = '/etc' %}

@ -0,0 +1,5 @@
{% calculate append = 'join', format = 'bind' -%}
options {
parameter-1 {{ variables.variable_1 }};
parameter-2 {{ variables.variable_2 }};
};

@ -0,0 +1 @@
{% calculate append = 'skip', action = 'install', path = '/etc' %}

@ -0,0 +1,5 @@
{% calculate append = 'join', format = 'bind' -%}
options {
parameter-1 {{ variables.variable_1 }};
parameter-2 {{ variables.variable_2 }};
};

@ -0,0 +1 @@
{% calculate append = 'skip', action = 'install', path = '/etc' %}

@ -0,0 +1 @@
{% calculate append = 'join', package = 'test-category/other-package' %}

@ -0,0 +1,5 @@
{% calculate append = 'join', format = 'bind' -%}
options {
parameter-1 {{ variables.variable_1 }};
parameter-2 {{ variables.variable_2 }};
};

@ -0,0 +1 @@
{% calculate append = 'join', package = 'test-category/test-package' %}

@ -0,0 +1,5 @@
{% calculate append = 'join', format = 'bind' -%}
options {
parameter-1 {{ variables.variable_1 }};
parameter-2 {{ variables.variable_2 }};
};

@ -0,0 +1 @@
{% calculate append = 'skip', action = 'install', path = '/etc' %}

@ -0,0 +1 @@
{% calculate append = 'join', package = 'test-category/other-package' %}

@ -0,0 +1,5 @@
{% calculate append = 'join', format = 'bind', merge = 'test-category/test-package' -%}
options {
parameter-1 {{ variables.variable_1 }};
parameter-2 {{ variables.variable_2 }};
};

@ -0,0 +1 @@
{% calculate append = 'join', package = 'test-category/test-package' %}

@ -0,0 +1,5 @@
{% calculate append = 'join', format = 'bind' -%}
options {
parameter-1 {{ variables.variable_1 }};
parameter-2 {{ variables.variable_2 }};
};

@ -0,0 +1 @@
{% calculate append = 'skip', package = 'test-category/test-package' %}

@ -0,0 +1,5 @@
{% calculate append = 'join', format = 'bind' -%}
options {
parameter-1 {{ variables.variable_1 }};
parameter-2 {{ variables.variable_2 }};
};

@ -0,0 +1 @@
{% calculate append = 'skip', package = 'test-category/test-package' %}

@ -0,0 +1,5 @@
{% calculate append = 'join', format = 'bind' -%}
options {
parameter-1 {{ variables.variable_1 }};
parameter-2 {{ variables.variable_2 }};
};

@ -0,0 +1 @@
{% calculate append = 'skip', action = 'install', path = '/etc' %}

@ -0,0 +1 @@
{% calculate append = 'join', package = 'test-category/other-package' %}

@ -0,0 +1,6 @@
{% calculate append = 'join', format = 'bind' %}
{% calculate merge = 'test-category/test-package, test-category/new-package' -%}
options {
parameter-1 {{ variables.variable_1 }};
parameter-2 {{ variables.variable_2 }};
};

@ -0,0 +1 @@
{% calculate append = 'join', package = 'test-category/test-package' %}

@ -0,0 +1,5 @@
{% calculate append = 'join', format = 'bind' -%}
options {
parameter-1 {{ variables.variable_1 }};
parameter-2 {{ variables.variable_2 }};
};

@ -0,0 +1 @@
{% calculate append = 'join', package = 'test-category/new-package' %}

@ -0,0 +1,5 @@
{% calculate append = 'join', format = 'bind' -%}
options {
parameter-1 {{ variables.variable_1 }};
parameter-2 {{ variables.variable_2 }};
};

@ -0,0 +1 @@
{% calculate append = 'skip', action = 'install', path = '/etc' %}

@ -0,0 +1 @@
{% calculate append = 'join', package = 'test-category/other-package' %}

@ -0,0 +1,6 @@
{% calculate append = 'join', format = 'bind' %}
{% calculate merge = 'test-category/new-package' -%}
options {
parameter-1 {{ variables.variable_1 }};
parameter-2 {{ variables.variable_2 }};
};

@ -0,0 +1 @@
{% calculate append = 'join', package = 'test-category/test-package' %}

@ -0,0 +1,5 @@
{% calculate append = 'join', format = 'bind' -%}
options {
parameter-1 {{ variables.variable_1 }};
parameter-2 {{ variables.variable_2 }};
};

@ -0,0 +1 @@
{% calculate append = 'join', package = 'test-category/new-package' %}

@ -0,0 +1,6 @@
{% calculate append = 'join', format = 'bind' -%}
{% calculate merge = 'test-category/test-package' -%}
options {
parameter-1 {{ variables.variable_1 }};
parameter-2 {{ variables.variable_2 }};
};

@ -0,0 +1 @@
{% calculate append = 'skip', action = 'install', path = '/etc' %}

@ -0,0 +1 @@
{% calculate append = 'join', package = 'test-category/test-package' %}

@ -0,0 +1,5 @@
{% calculate append = 'join', format = 'bind' -%}
options {
parameter-1 {{ variables.variable_1 }};
parameter-2 {{ variables.variable_2 }};
};

@ -0,0 +1 @@
{% calculate append = 'join', package = 'test-category/other-package' %}

@ -0,0 +1,6 @@
{% calculate append = 'join', format = 'bind' %}
{% calculate merge = 'test-category/test-package' %}
options {
parameter-1 {{ variables.variable_1 }};
parameter-2 {{ variables.variable_2 }};
};

@ -0,0 +1,6 @@
{% calculate append = 'join', format = 'bind' -%}
{% calculate package = 'test-category/test-package' -%}
options {
parameter-1 {{ variables.variable_1 }};
parameter-2 {{ variables.variable_2 }};
};

@ -6,3 +6,5 @@ obj /etc/file_4 c585be6f171462940b44af994a54040d 1593525253
dir /etc/dir_6
obj /etc/dir_6/file_0 c585be6f171462940b44af994a54040d 1593525253
obj /etc/dir_7/file_0 c585be6f171462940b44af994a54040d 1593525253
dir /etc/dir_12
obj /etc/dir_12/file_0 c585be6f171462940b44af994a54040d 1593525253

@ -0,0 +1,5 @@
dir /etc
dir /etc/dir_17
obj /etc/dir_17/file_0 c585be6f171462940b44af994a54040d 1593525253
dir /etc/dir_17/dir_0
obj /etc/dir_17/dir_0/file_0 c585be6f171462940b44af994a54040d 1593525253

@ -1,7 +1,14 @@
import pytest
import os
from calculate.vars.datavars import Namespace, Variable, CyclicVariableError, VariableError, ReadonlyVariable, ChoiceVariable, IntegerVariable, StringVariable, DefaultValue, TableVariable, HashVariable, VariableNotFoundError
from calculate.vars.vars_loader import NamespaceIniFiller, VariableLoader, ProfileFiller, NamespaceIniFillerStrict
from calculate.vars.datavars import Namespace, Variable, CyclicVariableError,\
VariableError, ReadonlyVariable,\
ChoiceVariable, IntegerVariable,\
StringVariable, DefaultValue,\
TableVariable, HashVariable,\
VariableNotFoundError
from calculate.vars.vars_loader import NamespaceIniFiller, VariableLoader,\
ProfileFiller, NamespaceIniFillerStrict
@pytest.mark.vars
class TestNamespace:
@ -45,7 +52,7 @@ class TestNamespace:
assert ns.os.linux.fullname.getValue() == "Calculate Linux Desktop"
assert ns.os.linux.root == ns
nsif.fill(ns,"""
nsif.fill(ns, """
[os][linux]
shortname = CLDX
""")
@ -59,32 +66,32 @@ class TestNamespace:
nsif.fill(ns, """
[os]
test = 123
[os]
test += 345
""")
assert ns.os.test.getValue() == "123,345"
nsif.fill(ns, """
[os]
test -= 123
""")
assert ns.os.test.getValue() == "345"
nsif.fill(ns, """
[os]
test += asdf,qwer,zxcv
""")
assert ns.os.test.getValue() == "345,asdf,qwer,zxcv"
nsif.fill(ns, """
[os]
test -= asdf,zxcv
""")
assert ns.os.test.getValue() == "345,qwer"
def test_fill_namespace_clear_namespaces(self):
@ -94,7 +101,7 @@ class TestNamespace:
[test][0]
dev = /dev/sda1
mount = swap
[test][1]
dev = /dev/sda2
mount = /
@ -104,19 +111,23 @@ class TestNamespace:
mount = /var/calculate
""")
assert [x.dev.getValue() for x in ns.test] == ['/dev/sda1', '/dev/sda2', '/dev/sda5']
assert [x.dev.getValue() for x in ns.test] == ['/dev/sda1',
'/dev/sda2',
'/dev/sda5']
nsif.fill(ns, """
[test][0]
dev = /dev/sdb1
mount = swap
[test][1]
dev = /dev/sdb2
mount = /
""")
assert [x.dev.getValue() for x in ns.test] == ['/dev/sdb1', '/dev/sdb2', '/dev/sda5']
assert [x.dev.getValue() for x in ns.test] == ['/dev/sdb1',
'/dev/sdb2',
'/dev/sda5']
nsif.fill(ns, """
[test][]
@ -124,13 +135,14 @@ class TestNamespace:
[test][0]
dev = /dev/sdb1
mount = swap
[test][1]
dev = /dev/sdb2
mount = /
""")
assert [x.dev.getValue() for x in ns.test] == ['/dev/sdb1', '/dev/sdb2']
assert [x.dev.getValue() for x in ns.test] == ['/dev/sdb1',
'/dev/sdb2']
def test_fill_namespace_strict_clear(self):
ns = Namespace()
@ -139,7 +151,7 @@ class TestNamespace:
[os][test][0]
dev = /dev/sda1
mount = swap
[os][test][1]
dev = /dev/sda2
mount = /
@ -165,15 +177,12 @@ class TestNamespace:
[custom][]
""")
assert [x for x in ns.custom ] == []
assert [x for x in ns.custom] == []
def test_get_namespace_attrs(self):
ns = Namespace()
os = Namespace("os")
os.addStringVariable("test","zxcv")
os.addStringVariable("test", "zxcv")
ns.addNamespace(os)
assert ns.os.test.getValue() == "zxcv"
@ -190,7 +199,7 @@ class TestNamespace:
return "A"
var = TestVar("test")
assert var.getValue() == "A"
def test_namespace_lookup(self):
@ -202,11 +211,11 @@ class TestNamespace:
os.addNamespace(linux)
os.addNamespace(device)
device1 = device.addNamespace()
device1.addStringVariable("dev","/dev/sda")
device1.addStringVariable("dev", "/dev/sda")
device2 = device.addNamespace()
device2.addStringVariable("dev","/dev/sdb")
device2.addStringVariable("dev", "/dev/sdb")
device3 = device.addNamespace()
device3.addStringVariable("dev","/dev/sdc")
device3.addStringVariable("dev", "/dev/sdc")
ns.addStringVariable("first", "first")
os.addStringVariable("second", "second")
@ -221,7 +230,7 @@ class TestNamespace:
assert linux.root.os.second.getValue() == "second"
with pytest.raises(VariableNotFoundError):
z = os.third
os.third
assert ns.os.device[0].dev.getValue() == "/dev/sda"
assert ns.os.device["0"].dev.getValue() == "/dev/sda"
@ -230,7 +239,8 @@ class TestNamespace:
assert ns.os.device.parent.second.getValue() == "second"
assert ns.os.device.parent.parent.os.second.getValue() == "second"
assert ns.os.device.parent.parent.parent.os.second.getValue() == "second"
assert ns.os.device.parent.parent.parent.os.second.\
getValue() == "second"
def test_variable_get_value_by_variable(self):
class TestVar1(Variable):
@ -278,7 +288,6 @@ class TestNamespace:
assert test1.counter == 3
def test_cyclic_variable(self):
class TestVar1(Variable):
def get(self):
return "%s,test1" % self.vars.test2.getValue(self)
@ -301,10 +310,12 @@ class TestNamespace:
ns.addVariable(test3)
with pytest.raises(CyclicVariableError) as e:
z = ns.test1.getValue()
assert e.value.queue[:-1] == ("test1","test2","test3")
ns.test1.getValue()
assert e.value.queue[:-1] == ("test1", "test2", "test3")
with pytest.raises(VariableError) as e:
z = ns.test1.getValue()
ns.test1.getValue()
def test_drop_invalidate_after_set_value(self):
class TestVar1(Variable):
@ -317,7 +328,7 @@ class TestNamespace:
class TestVar2(Variable):
def get(self):
return "ZZZZ"
test1 = TestVar1("test1")
test2 = TestVar2("test2")
ns = Namespace()
@ -325,7 +336,7 @@ class TestNamespace:
ns.addVariable(test2)
assert test1.getValue() == "ZZZZ,test1"
test1.setValue("VVVV")
assert test1.getValue() == "VVVV"
@ -344,9 +355,10 @@ class TestNamespace:
def test_change_invalidator_variable(self):
class VarTest(Variable):
counter = 0
def get(self):
self.counter += 1
if self.vars.ifvar.getValue(self):
return "%s,test1" % self.vars.vara.getValue(self)
else:
@ -384,7 +396,7 @@ class TestNamespace:
test1 = TestVar1("test1")
assert test1.getValue() == "test1"
with pytest.raises(VariableError) as e:
with pytest.raises(VariableError):
test1.setValue("test2")
assert test1.getValue() == "test1"
@ -399,7 +411,7 @@ class TestNamespace:
return ["test1", "test2"]
test1 = TestVar1("test1")
with pytest.raises(VariableError) as e:
with pytest.raises(VariableError):
test1.setValue("test3")
test1.setValue("test2")
assert test1.getValue() == "test2"
@ -409,7 +421,7 @@ class TestNamespace:
properties = [IntegerVariable]
test1 = TestVar1("test1")
with pytest.raises(VariableError) as e:
with pytest.raises(VariableError):
test1.setValue("test3")
test1.setValue("33")
assert test1.getValue() == 33
@ -447,7 +459,6 @@ class TestNamespace:
def get(self):
return "ZXC %s" % self.vars.test2.getCommentValue(self)
ns = Namespace()
test1 = TestVar1("test1")
assert test1.getValue() == "123"
@ -474,7 +485,7 @@ class TestNamespace:
properties = [ChoiceVariable]
def choice(self):
return ["test1","test2"]
return ["test1", "test2"]
class TestVar3(Variable):
properties = [ChoiceVariable]
@ -487,43 +498,54 @@ class TestNamespace:
test2 = TestVar2("test2")
test3 = TestVar3("test3")
with pytest.raises(VariableError) as e:
with pytest.raises(VariableError):
test1.choice()
with pytest.raises(VariableError) as e:
with pytest.raises(VariableError):
test1.choiceComment()
assert test2.choice() == ["test1","test2"]
assert test2.choiceComment() == [("test1","test1"),("test2","test2")]
assert test2.choice() == ["test1", "test2"]
assert test2.choiceComment() == [("test1", "test1"),
("test2", "test2")]
assert test3.choice() == ["test1","test2"]
assert test3.choiceComment() == [("test1","Test1"),("test2","Test2")]
assert test3.choice() == ["test1", "test2"]
assert test3.choiceComment() == [("test1", "Test1"),
("test2", "Test2")]
def test_loading_test_variable_module(self):
ns = Namespace()
vl = VariableLoader()
vl.fill(ns, "tests/vars/variables", "testvars")
assert ns.level.simple.getValue() == "simple value"
assert ns.level.uselocalsimple.getValue() == "Using simple value"
assert ns.level.usefullsimple.getValue() == "Using simple value"
with pytest.raises(VariableError) as e:
with pytest.raises(VariableError):
ns.level.badchoice.choice()
with pytest.raises(VariableError) as e:
with pytest.raises(VariableError):
ns.level.badchoice.choiceComment()
assert ns.level.simple_choice.choice() == ["/dev/sda1", "/dev/sda2", "/dev/sda3"]
assert ns.level.comment_choice.choice() == ["/dev/sda1", "/dev/sda2", "/dev/sda3"]
assert ns.level.comment_choice.choiceComment() == [("/dev/sda1", "SWAP"), ("/dev/sda2","ROOT"), ("/dev/sda3","DATA")]
assert ns.level.simple_choice.choice() == ["/dev/sda1",
"/dev/sda2",
"/dev/sda3"]
assert ns.level.comment_choice.choice() == ["/dev/sda1",
"/dev/sda2",
"/dev/sda3"]
assert ns.level.comment_choice.choiceComment() == [
("/dev/sda1", "SWAP"),
("/dev/sda2", "ROOT"),
("/dev/sda3", "DATA")]
ns.level.disks.setValue(["/dev/sda2", "/dev/sda1"])
assert ns.level.disks.getValue() == ["/dev/sda2", "/dev/sda1"]
assert ns.level.comment_choice.choice() == ["/dev/sda2", "/dev/sda1"]
assert ns.level.comment_choice.choiceComment() == [("/dev/sda2","ROOT"),
("/dev/sda1","SWAP")]
assert ns.level.comment_choice.choiceComment() == [
("/dev/sda2", "ROOT"),
("/dev/sda1", "SWAP")]
assert ns.level is not ns.level.level2.root
assert ns is ns.level.level2.root
@ -535,7 +557,6 @@ class TestNamespace:
vl = VariableLoader()
vl.fill(ns, "tests/vars/variables", "testvars")
assert ns.level.linux.getFullname() == "level.linux"
assert ns.level.linux.ver.fullname == "level.linux.ver"
@ -550,7 +571,7 @@ class TestNamespace:
# проверка установки значения hash переменной
ns.level.linux.ver.setValue("3.0")
assert ns.level.linux.ver.getValue() == "3.0"
# после установки хотя бы одного значения в hash переменной
# обновление остальных прекращаются до инвалидации (так как
# значения рассматриваются комплексно)
@ -558,7 +579,7 @@ class TestNamespace:
assert ns.level.linux.shortname.getValue() == "CLD"
# проверка попытки изменить readonly переменную
with pytest.raises(VariableError) as e:
with pytest.raises(VariableError):
ns.level.linux.shortname.setValue("CLDX")
# проверка сбора значения hash перемнной
@ -571,10 +592,9 @@ class TestNamespace:
# проверка обновления значения переменной, используеющей одно
# из значений hash переменной
assert ns.level.shortname_test.getValue() == "CLDG test"
ns.level.linux.shortname.setValue("CLDX",force=True)
ns.level.linux.shortname.setValue("CLDX", force=True)
assert ns.level.shortname_test.getValue() == "CLDX test"
def test_table_variable(self):
# table variable
ns = Namespace()
@ -583,77 +603,83 @@ class TestNamespace:
assert ns.level.device[0].dev.getValue() == "/dev/sda"
assert ns.level.device[1].dev.getValue() == "/dev/sdb"
assert ns.level.device.getValue() == [
{"dev":"/dev/sda",
"type":"hdd",
"name":"Samsung SSD"
},
{"dev":"/dev/sdb",
"type":"flash",
"name":"Transcend 64GB"
},
]
assert ns.level.device.getValue() == [{"dev": "/dev/sda",
"type": "hdd",
"name": "Samsung SSD"},
{"dev": "/dev/sdb",
"type": "flash",
"name": "Transcend 64GB"}]
assert ns.level.device[1].type.getValue() == "flash"
# проверка обновления списка пространства имён у табличной переменной
ns.level.devicelist.setValue(["/dev/sda","/dev/sdb","/dev/sdc"])
ns.level.devicelist.setValue(["/dev/sda", "/dev/sdb", "/dev/sdc"])
assert ns.level.device[2].type.getValue() == "usbhdd"
assert [x.type.getValue() for x in ns.level.device] == ["hdd", "flash", "usbhdd"]
assert [x.type.getValue() for x in ns.level.device] == ["hdd",
"flash",
"usbhdd"]
assert ns.level.device_child.getValue() == "hdd"
ns.level.devicelist.setValue(["/dev/sda","/dev/sdb"])
ns.level.devicelist.setValue(["/dev/sda", "/dev/sdb"])
ns.level.device[0].type.setValue("flash")
assert ns.level.device.getValue() == [
{"dev":"/dev/sda",
"type":"flash",
"name":"Samsung SSD"
},
{"dev":"/dev/sdb",
"type":"flash",
"name":"Transcend 64GB"
},
]
assert ns.level.device.getValue() == [{"dev": "/dev/sda",
"type": "flash",
"name": "Samsung SSD"},
{"dev": "/dev/sdb",
"type": "flash",
"name": "Transcend 64GB"}]
assert ns.level.device_child.getValue() == "flash"
# после установки хотя бы одного значения в table переменной
# обновление остальных прекращаются до инвалидации (так как
# значения рассматриваются комплексно)
ns.level.devicelist.setValue(["/dev/sda","/dev/sdb","/dev/sdc"])
assert [x.dev.getValue() for x in ns.level.device] == ["/dev/sda","/dev/sdb"]
ns.level.devicelist.setValue(["/dev/sda", "/dev/sdb", "/dev/sdc"])
assert [x.dev.getValue() for x in ns.level.device] == ["/dev/sda",
"/dev/sdb"]
ns.level.device.invalidate()
assert [x.dev.getValue() for x in ns.level.device] == ["/dev/sda","/dev/sdb","/dev/sdc"]
# проверить на повторное изменение, убедится, что _drop_child отрабатывает
ns.level.devicelist.setValue(["/dev/sda","/dev/sdb","/dev/sdc", "/dev/sdd"])
assert [x.dev.getValue() for x in ns.level.device] == ["/dev/sda",
"/dev/sdb",
"/dev/sdc"]
# проверить на повторное изменение, убедится, что _drop_child
# отрабатывает
ns.level.devicelist.setValue(["/dev/sda", "/dev/sdb",
"/dev/sdc", "/dev/sdd"])
ns.level.device.invalidate()
assert [x.dev.getValue() for x in ns.level.device] == ["/dev/sda","/dev/sdb","/dev/sdc", "/dev/sdd"]
assert [x.dev.getValue() for x in ns.level.device] == ["/dev/sda",
"/dev/sdb",
"/dev/sdc",
"/dev/sdd"]
def test_wrong_table_variable(self):
class Testtable(TableVariable):
class Data(TableVariable.Data):
def get(self):
return [{'dev':'123','name':'098'}]
return [{'dev': '123', 'name': '098'}]
ns = Namespace()
ns.addNamespace(Namespace("test"))
ns.test.addNamespace(Namespace("test2"))
error_message = ("Missed 'hash_vars' attribute for table "
"variable test.test2.testtable")
with pytest.raises(VariableError) as e:
ns.test.test2.addNamespace(Testtable("testtable", ns.test.test2))
assert str(e.value) == "Missed 'hash_vars' attribute for table variable test.test2.testtable"
assert str(e.value) == error_message
def test_wrong_hash_variable(self):
class Testhash(HashVariable):
class Data(HashVariable.Data):
def get(self):
return {'dev':'123','name':'098'}
return {'dev': '123', 'name': '098'}
ns = Namespace()
ns.addNamespace(Namespace("test"))
ns.test.addNamespace(Namespace("test2"))
error_message = ("Missed 'hash_vars' attribute for hash "
"variable test.test2.testhash")
with pytest.raises(VariableError) as e:
ns.test.test2.addNamespace(Testhash("testhash", ns.test.test2))
assert str(e.value) == "Missed 'hash_vars' attribute for hash variable test.test2.testhash"
assert str(e.value) == error_message
def test_namespace_iteration(self):
ns = Namespace()
@ -664,7 +690,7 @@ class TestNamespace:
ns2 = ns.addNamespace()
ns2.addStringVariable("test", "456")
assert [x.test.getValue() for x in ns] == ["123","234","456"]
assert [x.test.getValue() for x in ns] == ["123", "234", "456"]
def test_subnamespace(self):
ns = Namespace()
@ -679,20 +705,24 @@ class TestNamespace:
vl = VariableLoader()
vl.fill(ns, "tests/vars/variables", "testvars")
error_message = "Variable or namespace level.level3.myvar3 not found"
with pytest.raises(VariableError) as e:
z = ns.level.level3.myvar3.getValue()
assert str(e.value) == "Variable or namespace level.level3.myvar3 not found"
ns.level.level3.myvar3.getValue()
assert str(e.value) == error_message
# TODO: тест использует значения на конкретной машине
#def test_simple(self):
# ns = Namespace()
# vl = VariableLoader()
# vl.fill(ns, *VariableLoader.default())
# # нужно исправить тест, так
# # assert [x.name.getValue() for x in ns.os.gentoo.repositories] == ["gentoo","distros","calculate","custom"]
# assert ns.os.gentoo.make_profile.getValue() == "/etc/portage/make.profile"
# assert ns.os.gentoo.profile.path.getValue() == "/var/db/repos/distros/profiles/CLD/amd64/20"
# assert ns.os.gentoo.profile.name.getValue() == "distros:CLD/amd64/20"
# def test_simple(self):
# ns = Namespace()
# vl = VariableLoader()
# vl.fill(ns, *VariableLoader.default())
# # нужно исправить тест, так
# # assert [x.name.getValue() for x in ns.os.gentoo.repositories] ==\
# ["gentoo","distros","calculate","custom"]
# assert ns.os.gentoo.make_profile.getValue() ==\
# "/etc/portage/make.profile"
# assert ns.os.gentoo.profile.path.getValue() ==\
# "/var/db/repos/distros/profiles/CLD/amd64/20"
# assert ns.os.gentoo.profile.name.getValue() == "distros:CLD/amd64/20"
def test_profile_filler(self):
ns = Namespace()
@ -702,11 +732,15 @@ class TestNamespace:
class ProfileFillerTest(ProfileFiller):
def getRepositoryMap(self, ns):
curdir = os.getcwd()
return {
'distros': os.path.join(curdir,"tests/utils/gentoo/repos/distros"),
'calculate': os.path.join(curdir,"tests/utils/gentoo/repos/calculate"),
'gentoo': os.path.join(curdir,"tests/utils/gentoo/portage"),
}
return {'distros':
os.path.join(curdir,
"tests/utils/gentoo/repos/distros"),
'calculate':
os.path.join(curdir,
"tests/utils/gentoo/repos/calculate"),
'gentoo':
os.path.join(curdir,
"tests/utils/gentoo/portage")}
assert ns.os.hashvar.value1.getValue() == "test1"
assert ns.os.hashvar.value2.getValue() == "test2"
@ -740,12 +774,12 @@ class TestNamespace:
assert ns.os.tablevar[1].dev.getValue() == "/dev/sda2"
assert ns.os.tablevar[2].dev.getValue() == "/dev/sda5"
def test_fill_namespace_by_module(self):
ns = Namespace()
vl = VariableLoader()
vl.fill(ns, *VariableLoader.default())
assert "os" in ns
assert "config" in ns.os.gentoo
assert "main" in ns
assert ns.os.gentoo.config.getValue() is not None
pass
# ns = Namespace()
# vl = VariableLoader()
# vl.fill(ns, *VariableLoader.default())
# assert "os" in ns
# assert "config" in ns.os.gentoo
# assert "main" in ns
# assert ns.os.gentoo.config.getValue() is not None

Loading…
Cancel
Save