testomat.io управление авто тестамиНа одном проекте необходимо было настроить заглушки для разных окружений. Заглушки были одни и те же для разных окружений, а вот конфигурационные файлы менялись в зависимости от окружения.
В общем, это был полностью ручной процесс настройки. В какой-то момент мне поручили заниматься такой активностью “настройка заглушек для разных окружений” и конечно полностью ручной вариант обновления (а обновления были частыми) конфигурационных файлов под каждое окружения меня не устраивало. Я сделал небольшой python скрипт для копирования и замены данных в таких конфигурационных данных.
Вам хочу показать пример кода, как можно решать подобную задачу средствами python.
Что надо было?
Найди файлы для заглушек с одного окружения.
По патерну заменить данные в найденных файлах.
Файлы переместить в другое окружение.
Заресетить кэш на сервере.
Хотел сделать универсальный скрипт на все случаи жизни и чтобы можно было легко поддерживать, ну и наверное просто хотел покодить
, потому кода получилось много как для такой задачи.
Что сделал?
В общем я реализовал:
- Класс, который находит файлы, которые нужно заменять
- Класс, который работает с конфигом для скрипта
- Класс, который заменяет нужные данные по критериям указанным в конфиге
- Класс, который перемещает замененные файлы
- Класс, который ресетит кэш на сервере, чтобы заглушки обновились.
- Добавил нужные опции для запуска из консоли
Примечание: понятно что это узконаправленная задача и вряд ли кому-то пригодиться, но хочется делиться задачами и кодом, которые возникают в повседневной работе. А вдруг кому-то и пригодиться. Кода много потому, если у кого-то есть вопросы или замечания, пишите.
Буду рад, если Вы поделитесь подобными задачами и их реализациями
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()