mirror of
https://github.com/avatao-content/baseimage-tutorial-framework
synced 2024-12-23 02:11:55 +00:00
Implement git based fs snapshot provider
This commit is contained in:
parent
d5feba7076
commit
8e87025905
@ -9,3 +9,4 @@ from .history_monitor import HistoryMonitor, BashMonitor, GDBMonitor
|
||||
from .terminal_commands import TerminalCommands
|
||||
from .log_monitoring_event_handler import LogMonitoringEventHandler
|
||||
from .fsm_managing_event_handler import FSMManagingEventHandler
|
||||
from .snapshot_provider import SnapshotProvider
|
||||
|
54
lib/tfw/components/snapshot_provider.py
Normal file
54
lib/tfw/components/snapshot_provider.py
Normal file
@ -0,0 +1,54 @@
|
||||
# Copyright (C) 2018 Avatao.com Innovative Learning Kft.
|
||||
# All Rights Reserved. See LICENSE file for details.
|
||||
|
||||
from subprocess import run
|
||||
from getpass import getuser
|
||||
|
||||
|
||||
class SnapshotProvider:
|
||||
def __init__(self, directory, git_dir):
|
||||
author = f'{getuser()} via TFW {self.__class__.__name__}'
|
||||
self.gitenv = {
|
||||
'GIT_DIR': git_dir,
|
||||
'GIT_WORK_TREE': directory,
|
||||
'GIT_AUTHOR_NAME': author,
|
||||
'GIT_AUTHOR_EMAIL': '',
|
||||
'GIT_COMMITTER_NAME': author,
|
||||
'GIT_COMMITTER_EMAIL': ''
|
||||
}
|
||||
|
||||
def init_repo(self):
|
||||
self._run(('git', 'init'))
|
||||
|
||||
def take_snapshot(self):
|
||||
self._run(('git', 'add', '-A'))
|
||||
self._run(('git', 'commit', '-m', 'Snapshot'))
|
||||
|
||||
def restore_snapshot(self, date):
|
||||
commit = self._get_commit_from_timestamp(date)
|
||||
self._checkout_commit(commit)
|
||||
|
||||
def _get_commit_from_timestamp(self, date):
|
||||
return self._get_stdout((
|
||||
'git', 'rev-list',
|
||||
'--date=iso',
|
||||
'-n', '1',
|
||||
f'--before="{date.isoformat()}"',
|
||||
'master'
|
||||
))
|
||||
|
||||
def _checkout_commit(self, commit):
|
||||
self._run((
|
||||
'git', 'checkout',
|
||||
commit
|
||||
))
|
||||
|
||||
def _get_stdout(self, *args, **kwargs):
|
||||
kwargs['capture_output'] = True
|
||||
stdout_bytes = self._run(*args, **kwargs).stdout
|
||||
return stdout_bytes.decode().rstrip('\n')
|
||||
|
||||
def _run(self, *args, **kwargs):
|
||||
if 'env' not in kwargs:
|
||||
kwargs['env'] = self.gitenv
|
||||
return run(*args, **kwargs)
|
Loading…
Reference in New Issue
Block a user