Есть отличная удаленная работа для php+codeception+jenkins+allure+docker спецов. 100% remote! Присоединиться к проекту

Работа с конфигурационными файлами на разных окружениях, пример скрипта на Python


(Mykhailo Poliarush) #1

На одном проекте необходимо было настроить заглушки для разных окружений. Заглушки были одни и те же для разных окружений, а вот конфигурационные файлы менялись в зависимости от окружения.

В общем, это был полностью ручной процесс настройки. В какой-то момент мне поручили заниматься такой активностью “настройка заглушек для разных окружений” и конечно полностью ручной вариант обновления (а обновления были частыми) конфигурационных файлов под каждое окружения меня не устраивало. Я сделал небольшой python скрипт для копирования и замены данных в таких конфигурационных данных.

Вам хочу показать пример кода, как можно решать подобную задачу средствами python.

Что надо было?

Найди файлы для заглушек с одного окружения.
По патерну заменить данные в найденных файлах.
Файлы переместить в другое окружение.
Заресетить кэш на сервере.

Хотел сделать универсальный скрипт на все случаи жизни и чтобы можно было легко поддерживать, ну и наверное просто хотел покодить :smile: , потому кода получилось много как для такой задачи.

Что сделал?

В общем я реализовал:

  1. Класс, который находит файлы, которые нужно заменять
  2. Класс, который работает с конфигом для скрипта
  3. Класс, который заменяет нужные данные по критериям указанным в конфиге
  4. Класс, который перемещает замененные файлы
  5. Класс, который ресетит кэш на сервере, чтобы заглушки обновились.
  6. Добавил нужные опции для запуска из консоли

Примечание: понятно что это узконаправленная задача и вряд ли кому-то пригодиться, но хочется делиться задачами и кодом, которые возникают в повседневной работе. А вдруг кому-то и пригодиться. Кода много потому, если у кого-то есть вопросы или замечания, пишите.

Буду рад, если Вы поделитесь подобными задачами и их реализациями

config.ini

[FILE_PATTERNS] 
# * and ? can be used to file patterns
# * - match sequence of symbols
# ? - match only one symbol
async: *asyncResponse.xml
conf: *configuration.xml
sync: *syncResponse.xml
info: *info*.txt

[STUB_SERVER]
server: host.server.com
ssh_port: 22 
http_port: 5555
login: account
pass: password
server_path: /IntegrationServer/config/stubs

[WM_IS_CREDENTIALS]
login: account 
pass: password

[DEV]
cop_host: host.corp.server.com
cop_port: 6610
tom_host: host.corp.server.com
tom_port: 6620
cop_endpoints_to: billing, cap
tom_endpoints_to: logistics, provident

[DEV2]
cop_host: host.corp.server.com
cop_port: 6630
tom_host: host.corp.server.com
tom_port: 6640
cop_endpoints_to: billing, vasep
tom_endpoints_to: orderResource

update_stub.py

"""python %prog [options]

Description:
    This is a program to move STUBs files between environments
    All environment configurations you can find in file config.ini.
    You can edit config.ini up to your needs.
    All you should know that config.ini should exist before program running.
Example:
    python %prog -from STUBS_TEMPLATES -to SYS2
Defaults:
    If you want to move files from STUBS_TEMPLATES to SYS2 env
    you can run proram without any options passed.
config.ini:
    It's just simple file that holds configuration information about environments.
    Format:
    [<name of env as it's in file system>]
    <option>: <value>
"""

import os
import fnmatch
import sys
import re
import urllib2
from optparse import OptionParser
from ConfigParser import ConfigParser
from shutil import copy2


def any(iterable):
    for element in iterable:
        if element:
            return True
    return False


class EnvCofigurationManager:
    """ store and handle environment data """
    _env = {}

    def __init__(self, path="config.ini"):
        config = ConfigParser()
        config.read([path])
        for section in config.sections():
            self._env[section] = dict(config.items(section))

    def __getitem__(self, key):
        if key not in self._env:
            raise Exception("'%s' key is not found in environment configuration from config.ini" % key)
        return self._env[key]

    def __contains__(self, env):
        return env in self._env

try:
    env = EnvCofigurationManager()
except Exception, e:
    print "Can't read configuration file config.ini"
    print e
    sys.exit(1)


class StatusManager:

    """ manage status of operation """
    status = True

    def __eq__(self, value):
        return self.status


class FileFinder(StatusManager):

    """ file neccessary data to be working on """
    def __init__(self, dir=os.curdir, from_env=None):
        print "\n\nFOUND FILES TO PROCESS:"
        self._files = []
        self._dir_to_lookup = dir
        self._from_env = from_env
        self._find_files()

    def _find_files(self):
        """ find files by generator """
        for file_path in self._locate_files_by_pattern():
            self._files.append(file_path)
        # FIX: there is duplication in generator, as for now remove it by set convertion
        self._files = list(set(self._files))
        self._print_files_path()

    def _locate_files_by_pattern(self):
        """ go thru nested folder and match files by pattern """
        if self._from_env:
            # to refactor code
            for path, dirs, files in os.walk(self._dir_to_lookup):
                if self._from_env in dirs:
                    _directory = os.path.abspath(os.path.join(path, self._from_env))
                    break
        else:
            _directory = self._dir_to_lookup
        for path, dirs, files in os.walk(_directory):
            for pattern in env["FILE_PATTERNS"].values():
                for filename in fnmatch.filter(files, pattern):
                    file_path = os.path.abspath(os.path.join(path, filename))
                    yield file_path

    def _print_files_path(self):
        for i, file_path in enumerate(self._files):
            print "%d. %s" % (i + 1, file_path)

    @property
    def path(self):
        return self._files

    def __gt__(self, value):
        return len(self._files) > value

    def __repr__(self):
        return "Currently found %d files to process in directory %s" % (len(self._files), self._dir_to_lookup)


class FileReplacement(StatusManager):

    """ replace files with necessary data """
    def __init__(self, files, **env_data):
        print "\n\nREPLACEMENT:"
        self._files_to_replace = files
        self._from = env_data["from_env"]
        self._to = env_data["to_env"]
        self._match_url = re.compile(r"(<url>https?://)((?:\w+\.?)+):(\d+)(/(?:.*?)</url>)")

    def replace_endpoint_host_and_port(self):
        """ replace host and port on endpoint """
        _cop_folders = map(lambda x: x.strip(), env[self._to]["cop_endpoints_to"].split(","))
        _tom_folders = map(lambda x: x.strip(), env[self._to]["tom_endpoints_to"].split(","))
        i = 1
        for file in self._files_to_replace:
            try:
                file_object = open(file, "r+")
                file_content = file_object.read()
                _matched_url = self._match_url.search(file_content)
                if _matched_url:
                    file_object.seek(0)
                    file_object.truncate()
                    if any(filter(lambda x: x in file, _tom_folders)):
                        to_what_replace = r"\1%s:%s\4" % (env[self._to]["tom_host"], env[self._to]["tom_port"])
                    elif any(filter(lambda x: x in file, _cop_folders)):
                        to_what_replace = r"\1%s:%s\4" % (env[self._to]["cop_host"], env[self._to]["cop_port"])
                    else:
                        pass
                    new_content = self._match_url.sub(to_what_replace, file_content)
                    file_object.write(new_content)
                    print "%d.\t" % i + file + " is changed"
                    i += 1
            except IOError:
                self.status = False
                print "Can't read file %s" % file
        return True

    def replace_endpoint_url(self, url):
        """ replace whole url on endpoint """
        pass


class CashServerManagement(StatusManager):

    """ reset cache on server """
    def __init__(self, wm_is):
        print "\n\nRESET CACHE ON SERVER:"
        self._wm_is = wm_is
        self._wm_is_url = "http://%s:%s/WmRoot/stats-services.dsp?action=resetcache" % (wm_is["cop_host"], wm_is["cop_port"])
        self._login = env["WM_IS_CREDENTIALS"]["login"]
        self._pass = env["WM_IS_CREDENTIALS"]["pass"]
        self._reset_cache()

    def _reset_cache(self):
        req = urllib2.Request(self._wm_is_url)

        password_manager = urllib2.HTTPPasswordMgrWithDefaultRealm()
        password_manager.add_password(None, self._wm_is_url, self._login, self._pass)

        auth_manager = urllib2.HTTPBasicAuthHandler(password_manager)
        opener = urllib2.build_opener(auth_manager)
        urllib2.install_opener(opener)

        try:
            handler = urllib2.urlopen(req)
            if "server cache cleared" not in handler.read().lower():
                self.status = False
        except urllib2.HTTPError, e:
            print e
            self.status = False
        print "done"


class RemoteFileManager(StatusManager):

    def __init__(self, from_env, to_env, files):
        print "\n\nFILE COPY:"
        self._from_env, self._to_env, self._files = from_env, to_env, files
        self._copy_files()

    def _copy_files(self):
        number_of_copied_files = 0
        for file in self._files:
            from_file = file
            to_file = env["STUB_SERVER"]["server_path"] + os.sep + self._to_env + from_file.split(self._from_env)[1]
            folder_to_copy = os.path.dirname(to_file)
            if not os.path.exists(folder_to_copy):
                os.makedirs(folder_to_copy)
            try:
                copy2(from_file, to_file)
                number_of_copied_files += 1
                print "%d. %s" % (
                    number_of_copied_files,
                    to_file)
            except Exception, e:
                print "i'm in exception", e
                self.status = False
        print "Copied %d out of %d files from %s to %s" % (number_of_copied_files, len(self._files), self._from_env, self._to_env)


if __name__ == "__main__":
    usage = __doc__
    parser = OptionParser(usage=usage)
    parser.add_option("-f", "--from", default="STUBS_TEMPLATES", action="store", dest="from_env",
                      help="define from what environment you want to copy files")
    parser.add_option("-t", "--to", default="SYS2", action="store", dest="to_env",
                      help="define to what environment you want to copy files")
    parser.add_option("-d", "--dir", default=".", action="store", dest="dir",
                      help="directory where to look up files")
    parser.add_option("-u", "--update", action="store", dest="update",
                      help="update env. stubs as per latest from repository")
    (options, args) = parser.parse_args()

    assert options.from_env in env, "%s is not supported or you did a mistake" % options.from_env
    assert options.to_env in env, "%s is not supported or you did a mistake" % options.to_env

    if options.update:
        assert options.update in env, "%s is not supported or you did a mistake" % options.update

    if options.update:
        files = FileFinder(options.dir, options.update)
        assert files > 0, "No files found to process"
        assert RemoteFileManager(options.update, options.update, files.path)
        assert CashServerManagement(env[options.update]), "Can't reset cache on server %r" % env[options.to_env]
    elif options.dir and options.from_env:
        files = FileFinder(options.dir, options.from_env)
        assert files > 0, "No files found to process"
        config_files = fnmatch.filter(files.path, env["FILE_PATTERNS"]["conf"])
        replacer = FileReplacement(config_files, from_env=options.from_env, to_env=options.to_env)
        assert replacer.replace_endpoint_host_and_port()
        assert RemoteFileManager(options.from_env, options.to_env, files.path)
        assert CashServerManagement(env[options.to_env]), "Can't reset cache on server %r" % env[options.to_env]
    else:
        print "No options are passed to program"
        sys.exit()

Помогите разобраться с циклом чтобы написать веб-драйвер тест