KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use _resort method in autotest
Best Python code snippet using autotest_python
mysql.py
Source:mysql.py 
1import configparser2import datetime3import os4import pyAesCrypt5import re6import resource7import shutil8import subprocess9import sys10import tarfile11import tempfile12import time13class DeletionMeta():14 def __init__(self):15 self._candidates = {}16 def insert(self, candidate):17 candidate.inject(self._candidates)18 return self19 def candidateForBackup(self, backup):20 return DeletionCandidate.byBackup(self._candidates, backup)21class DeletionCandidate():22 def __init__(self, meta):23 self._expirationTime = None24 self._meta = meta25 def remove(self, resort):26 self._backup.remove(resort)27 return self28 @classmethod29 def byBackup(cls, dict, backup):30 return dict[backup.getName()]31 def inject(self, dict):32 dict[self._backup.getName()] = self33 return self34 def getName(self):35 return self._backup.getName()36 def parse(self, backup):37 self._backup = backup38 self._meta.insert(self)39 return self40 def rule(self, rule):41 tagName, validTime = rule.split(':')42 if not self._backup.hasTag(tagName):43 return self44 unit, amount = self.__parseRuleIntoUnitAmonut(validTime)45 interval = self.__parseExpirationInterval(unit, amount)46 newExpirationTime = self._backup.getCreationDate() + interval47 if self.isLater(newExpirationTime):48 self._expirationTime = newExpirationTime49 return self50 def __parseRuleIntoUnitAmonut(self, validTime):51 result = re.match('([0-9]+)(.*)', validTime)52 if not result:53 raise ValueError(validTime+' could not be parsed as valid time')54 amount = int(result.group(1))55 unit = result.group(2)56 return unit, amount57 def __parseExpirationInterval(self, unit, amount):58 units = {59 'day': self.daysDelta,60 'days': self.daysDelta,61 'week': self.weeksDelta,62 'weeks': self.weeksDelta,63 }64 try:65 interval = units[unit](amount)66 except KeyError:67 validUnits = ','.join( units.keys() )68 raise KeyError(unit+' is not a valid unit. Accepted values: '+validUnits)69 return interval70 def isLater(self, newExpirationTime):71 if not self._expirationTime:72 return True73 isLater = newExpirationTime > self._expirationTime74 if isLater:75 return True76 return False77 def daysDelta(self, amount):78 return datetime.timedelta(days=amount)79 def weeksDelta(self, amount):80 return datetime.timedelta(weeks=amount)81 def orphaned(self):82 if self._backup.isFull():83 return False84 parents = self._backup.getParents()85 for parent in parents:86 parentCandidate = self._meta.candidateForBackup(parent)87 if not parentCandidate.shouldDelete():88 return False89 return True90 def expired(self):91 if not self._expirationTime:92 return False93 return self._expirationTime < datetime.datetime.now()94 def shouldDelete(self):95 if self.expired():96 return True97 if self.orphaned():98 return True99 return False100 def printExpiration(self):101 self._backup.print(suffix=' expires at '+str(self._expirationTime) )102 def print(self):103 if not self.shouldDelete():104 return105 if self.expired():106 self._backup.print(suffix=' has expired at '+str(self._expirationTime))107 if self.orphaned():108 self._backup.print(suffix=' will be orphaned after the pruning')109 def getCreationDate(self):110 return self._backup.getCreationDate()111class Finder():112 def backups(self, backups):113 self.backups = backups114 return self115 def parameters(self, parameters):116 self._parameters = parameters117 return self118class Sorter():119 def sort(self, backups):120 return sorted(backups, key=lambda backup: backup.getCreationDate())121class LatestTagFinder(Finder):122 def __init__(self, sorter=Sorter()):123 self._sorter = sorter124 self._parameters = []125 def find(self, backups):126 if len(self._parameters) == 0:127 raise KeyError("Missing tag to search")128 requiredTag = self._parameters[0]129 backupsWithTags = []130 for backup in backups:131 if backup.hasTag(requiredTag):132 backupsWithTags.append(backup)133 sortedBackups = self._sorter.sort(backupsWithTags)134 return sortedBackups[-1]135class LatestFinder(Finder):136 def __init__(self, sorter=Sorter()):137 self._sorter = sorter138 def find(self, backups):139 sortedBackups = self._sorter.sort(backups)140 return sortedBackups[-1]141class LatestFullFinder(Finder):142 def __init__(self, sorter=Sorter()):143 self._sorter = sorter144 def find(self, backups):145 fullBackups = []146 for backup in backups:147 if not backup.isFull():148 continue149 fullBackups.append(backup)150 sortedFullBackups = self._sorter.sort(fullBackups)151 return sortedFullBackups[-1]152class BackupNotFoundException(Exception):153 pass154class MySQLBackupMeta:155 def __init__(self):156 self.byEndingPoint = {}157 self.byStartingPoint = {}158 def addByEndingPoint(self, endingPoint,backup):159 try:160 self.byEndingPoint[endingPoint].append(backup)161 except KeyError:162 self.byEndingPoint[endingPoint] = [ backup ]163 return self164 def addByStartingPoint(self, startingPoint,backup):165 try:166 self.byStartingPoint[startingPoint].append(backup)167 except KeyError:168 self.byStartingPoint[startingPoint] = [ backup ]169 return self170 def parents(self, childStartingPoint):171 try:172 return self.byEndingPoint[childStartingPoint]173 except KeyError:174 return None175 def children(self, parentEndingPoint):176 try:177 return self.byStartingPoint[parentEndingPoint]178 except KeyError:179 return []180class MySQLBackup:181 def __init__(self, sorter=Sorter()):182 self._name = ''183 self._sorter = sorter184 self._tags = []185 def remove(self, resort):186 resort.remove(self._name)187 return self188 def getName(self):189 return self._name190 def addTag(self, tag):191 if tag in self._tags:192 return self193 self._tags.append(tag)194 return self195 def mysql(self, mysql):196 self._mysql = mysql197 return self198 def name(self, name):199 self._name = name200 return self201 def meta(self, meta):202 self._meta = meta203 return self204 def hasTag(self, tag):205 return tag in self._tags206 def writeIni(self, filePath):207 config = configparser.ConfigParser()208 config['main'] = {}209 config['main']['tags'] = ','.join(self._tags)210 with open(filePath, 'w') as file:211 config.write(file)212 def parseIni(self, extraIni):213 config = configparser.ConfigParser()214 config.read_string(extraIni)215 self._tags = config['main']['tags'].split(',')216 return self217 def parseInfo(self, infoContent):218 dummyHeader = '[main]\n'219 iniFile = dummyHeader+infoContent220 xtrabackupInfoContent = configparser.ConfigParser()221 xtrabackupInfoContent.read_string(iniFile)222 isFullBackup = False223 startingPoint = int(xtrabackupInfoContent['main']['innodb_from_lsn'])224 if startingPoint == 0:225 isFullBackup = True226 self._full = isFullBackup227 self._startingPoint = startingPoint228 endingPoint = int(xtrabackupInfoContent['main']['innodb_to_lsn'])229 self._endingPoint = endingPoint230 self._meta.addByEndingPoint(endingPoint, self)231 self._meta.addByStartingPoint(startingPoint, self)232 return self233 def isNamed(self, name):234 return self._name == name235 def isFull(self):236 return self._full237 def print(self, indent = 0, prefix='', suffix=''):238 tags = self._tags239 tagList = ','.join(self._tags)240 tagInfo = ' tags:'+tagList241 print( (' ' * indent) + prefix + self._name + tagInfo + suffix)242 def printRecursive(self, indent = 0):243 self.print(indent)244 sortedChildren = self._sorter.sort( self.getChildren() )245 for child in sortedChildren:246 if child is self:247 continue248 if not child.isClosestRelative(self):249 continue250 child.printRecursive(indent + 2)251 def isBefore(self, other):252 return self.getCreationDate() < other.getCreationDate()253 def getClosestRelative(self):254 children = self.getChildren()255 sortedChildren = self._sorter.sort(children)256 return sortedChildren[0]257 def isClosestRelative(self, other):258 try:259 closestRelative = other.getClosestRelative()260 return closestRelative == self261 except IndexError:262 return False263 def getCreationDate(self):264 return datetime.datetime.strptime(self._name, '%Y-%m-%d_%H-%M')265 def getHistory(self):266 history = [self]267 parent = self.getParent()268 while parent:269 history.append(parent)270 parent = parent.getParent()271 history.reverse()272 fullBackup = history.pop(0)273 return [ fullBackup, history ]274 def isParent(self, backup):275 if backup in self.getParents():276 return True277 for parent in self.getParents():278 if parent.isParent(backup):279 return True280 return False281 def getParent(self):282 if self._full:283 return None284 parents = self._meta.parents(self._startingPoint)285 return self.__skipEmptyParents(parents)[0]286 def getParents(self):287 if self._full:288 return []289 parents = self._meta.parents(self._startingPoint)290 parents = self.__cantBeOwnParent(parents)291 return self.__parentsMustBeBefore(parents)292 def __cantBeOwnParent(self, parents):293 return list( filter(lambda backup: backup._name != self._name, parents) )294 def __skipEmptyParents(self, parents):295 return list( filter(lambda backup: not backup.isEmpty(), parents) )296 def __parentsMustBeBefore(self, parents):297 return list( filter(lambda backup: backup.isBefore(self), parents) )298 def isEmpty(self):299 return self._startingPoint == self._endingPoint300 def getChildren(self):301 children = self._meta.children(self._endingPoint)302 withoutSelf = self.__cantBeOwnChild(children)303 return self.__childCantBeBefore(withoutSelf)304 def __cantBeOwnChild(self, children):305 return list( filter(lambda backup: backup._name != self._name, children) )306 def __childCantBeBefore(self, children):307 return list( filter(lambda backup: self.isBefore(backup), children ) )308 def inject(self, backupList):309 backupList.append(self)310 return self311 def setTimestamp(self, gauge):312 timestamp = self.getCreationDate().timestamp()313 gauge.set(timestamp)314 return self315class MySQL:316 def __init__(self, sorter=Sorter()):317 self._mysqlHost = os.environ.get('MYSQL_HOST', 'mysql')318 self._mysqlPort = os.environ.get('MYSQL_PORT', '3306')319 self._mysqlUsername = os.environ.get('MYSQL_USERNAME', 'backup')320 self._mysqlPassword = os.environ.get('MYSQL_PASSWORD')321 self._mysqlDatadir = os.environ.get('MYSQL_DATADIR', False)322 self._backupCommand = os.environ.get('MYSQL_COMMAND', 'mariabackup')323 self._bufferSize = int(os.environ.get('MYSQL_ENC_BUFSIZE', 64 * 1024))324 self._password = os.environ.get('MYSQL_ENC_PASSWORD')325 self._assetBase = os.path.dirname(os.path.realpath(__file__))+'/assets'326 self._tags = []327 self._sorter = sorter328 self._dryRun = False329 self._specialNames = {330 'latest-full-backup': LatestFullFinder(),331 'latest-backup': LatestFinder(),332 'latest-tag': LatestTagFinder(),333 }334 def setOutput(self, output):335 self.output = output336 def dataDir(self, dataDir):337 self._mysqlDatadir = dataDir338 return self339 def dryRun(self, dryRun):340 self._dryRun = dryRun341 return self342 def resort(self, resort):343 self._resort = resort344 return self345 def pruneBackups(self, rules):346 candidates = self.__buildDeletionCanidates(rules)347 sortedCandidates = self._sorter.sort(candidates)348 for candidate in sortedCandidates:349 candidate.print()350 if self._dryRun:351 print("Dry run - not executing")352 return353 for candidate in candidates:354 if not candidate.shouldDelete():355 continue356 candidate.remove( self._resort.adapter('mysql') )357 def __buildDeletionCanidates(self, rules):358 candidates = []359 meta = DeletionMeta()360 backups = self.list()361 for backup in backups:362 candidate = DeletionCandidate(meta).parse(backup)363 for rule in rules:364 candidate.rule(rule)365 candidates.append(candidate)366 return candidates367 def incrementalBackup(self, name, parentName):368 parent = self.find(parentName)369 print('Basing backup on:')370 parent.print()371 return self.__backup(name, parent._endingPoint)372 def fullBackup(self, name):373 return self.__backup(name)374 def tags(self, tags):375 self._tags = tags376 return self377 def __backup(self, name, baseLsn = None):378 fileLimit = self.__getFileLimit()379 with tempfile.TemporaryDirectory() as tempDirectory:380 backupDirectory = tempDirectory+'/backup'381 os.mkdir(backupDirectory)382 command = [383 'mariabackup',384 '--backup',385 '--target-dir='+backupDirectory,386 '--host='+self._mysqlHost,387 '--user='+self._mysqlUsername,388 '--password='+self._mysqlPassword,389 '--port='+self._mysqlPort,390 ]391 if self._mysqlDatadir:392 command.append('--datadir='+self._mysqlDatadir)393 if baseLsn:394 command.append('--incremental-lsn='+str(baseLsn))395 completedProcess = subprocess.run(command, check=True)396 self.__extractBackupInfo(backupDirectory, tempDirectory)397 tarFile = self.__compressBackup(backupDirectory, tempDirectory)398 self.__encryptBackup(tarFile)399 newBackup = MySQLBackup()400 for tag in self._tags:401 newBackup.addTag(tag)402 newBackup.writeIni(tempDirectory+'/cloudbackup.ini')403 self.__uploadBackup(tempDirectory, name)404 return self405 def __extractBackupInfo(self, backupDirectory, tempDirectory):406 shutil.copy(backupDirectory+'/xtrabackup_info', tempDirectory+'/xtrabackup_info')407 def __compressBackup(self, backupDirectory, tempDirectory):408 tarFilePath = tempDirectory+'/backup.tar.bz2'409 tar = tarfile.open(tarFilePath, 'w:bz2')410 tar.add(backupDirectory, 'backup')411 tar.close()412 shutil.rmtree(backupDirectory)413 return tarFilePath414 def __encryptBackup(self, tarFilePath):415 pyAesCrypt.encryptFile(tarFilePath, tarFilePath+'.aes', self._password, self._bufferSize)416 os.remove(tarFilePath)417 def __uploadBackup(self, backupDirectory, name):418 self._resort.adapter('mysql').upload(backupDirectory, name)419 def getSpecialBackups(self):420 availableBackups = self.list()421 specialBackups = {}422 for specialName in self._specialNames:423 specialNameFinder = self._specialNames[specialName]424 try:425 specialBackups[specialName] = specialNameFinder.find(availableBackups)426 except IndexError:427 pass428 except KeyError:429 pass430 return specialBackups431 def find(self, name):432 availableBackups = self.list()433 specialParameters = name.split(':')434 specialName = specialParameters.pop(0)435 if specialName in self._specialNames.keys():436 specialNameFinder = self._specialNames[specialName]437 return specialNameFinder.parameters(specialParameters).find(availableBackups)438 for backup in availableBackups:439 if backup.isNamed(name):440 return backup441 raise BackupNotFoundException()442 def restore(self, name, dataDir, port=8080):443 dataDirAbsolute = os.path.abspath(dataDir)444 backup = self.find(name)445 fullBackup, history = backup.getHistory()446 self.output.info("Full Backup: "+fullBackup.getName())447 self.output.info("Backup history:")448 for backup in history:449 self.output.info("- "+backup.getName() )450 self.output.info("Restoring Full Backup")451 self.__restoreFullBackup(fullBackup, dataDirAbsolute)452 self.output.info("Restoring Incremental Backups")453 self.__restoreIncrementalBackups(history, dataDirAbsolute)454 self.__copyDockerCompose(dataDirAbsolute+'/backup/docker-compose.yml', port)455 def __restoreFullBackup(self, backup, dataDir):456 self.__downloadAndExtract(backup, dataDir)457 completedProcess = subprocess.run([458 'mariabackup',459 '--prepare',460 '--target-dir='+dataDir+'/backup',461 ], check=True)462 def __restoreIncrementalBackups(self, backups, dataDir):463 for backup in backups:464 self.__restoreIncrementalBackup(backup, dataDir)465 def __restoreIncrementalBackup(self, backup, dataDir):466 with tempfile.TemporaryDirectory(dir=dataDir) as tempDir:467 self.output.info("Downloading "+backup.getName())468 self.__downloadAndExtract(backup, tempDir)469 completedProcess = subprocess.run([470 'mariabackup',471 '--prepare',472 '--target-dir='+dataDir+'/backup',473 '--incremental-dir='+tempDir+'/backup'474 ], check=True)475 def __downloadAndExtract(self, backup, targetDirectory):476 self._resort.adapter('mysql').download(backup._name, targetDirectory)477 tarFilePath = targetDirectory+'/backup.tar.bz2'478 self.__decryptBackup(tarFilePath)479 self.__unpackBackup(tarFilePath, targetDirectory)480 def __decryptBackup(self, tarFilePath):481 encryptedPath = tarFilePath+'.aes'482 pyAesCrypt.decryptFile(encryptedPath, tarFilePath, self._password, self._bufferSize)483 os.remove(encryptedPath)484 def __unpackBackup(self, tarFilePath, dataDir):485 tar = tarfile.open(tarFilePath, 'r:bz2')486 tar.extractall(dataDir)487 os.remove(tarFilePath)488 def __copyDockerCompose(self, target, port):489 with open(self._assetBase+'/docker-compose.yml', 'r') as file:490 dockerCompose = file.read()491 dockerCompose = dockerCompose.replace('%%PORT%%', str(port))492 with open(target, 'w') as file:493 file.write(dockerCompose)494 def __getFileLimit(self):495 limits = resource.getrlimit(resource.RLIMIT_NOFILE)496 return limits[1]497 def list(self):498 meta = MySQLBackupMeta()499 backups = []500 for directory in self._resort.adapter('mysql').listFolders():501 backup = self.__parseBackup(directory, meta)502 backup.inject(backups)503 sortedBackups = self._sorter.sort(backups)504 return sortedBackups505 def __parseBackup(self, directory, meta):506 backupName = os.path.basename(directory)507 infoContent = self._resort.adapter('mysql').fileContent(directory+'/xtrabackup_info').decode('utf-8')508 cloudbackupIni = self._resort.adapter('mysql').fileContent(directory+'/cloudbackup.ini').decode('utf-8')509 return MySQLBackup() \510 .name(backupName).meta(meta) \511 .parseInfo(infoContent) \512 .parseIni(cloudbackupIni)513 def scrape(self, gauge):514 availableBackups = self.list()515 self.__scrapeLatestBackup(gauge, availableBackups)516 self.__scrapeLatestFullBackup(gauge, availableBackups)517 return self518 def __scrapeLatestBackup(self, gauge, backups):519 try:520 backup = self._specialNames['latest-backup'].find(backups)521 backup.setTimestamp( gauge.labels(self._resort._name, 'Either') )522 except IndexError:523 gauge.labels(self._resort._name, 'Either').set(0)524 def __scrapeLatestFullBackup(self, gauge, backups):525 try:526 fullBackup = self._specialNames['latest-full-backup'].find(backups)527 fullBackup.setTimestamp( gauge.labels(self._resort._name, 'True') )528 except IndexError:...
test_board.py
Source:test_board.py
1#!/usr/bin/python32import unittest3import board4# Deck fixtures5unshuffled = [*range(0,52)]6reversed = [*range(0,52)]7reversed.reverse()8no_aces = board.parseDeck("""99H KD 8H JD AC 9S 7C 6D10JH 3H 7S TC AH 2C 6C 4H11KH AD TH 2D 3S QC JS 4S125D TD 8C TS 6H JC 8S 7D13QS QD 4D 3C AS 5H 7H QH145C KC 8D 9C 2H 3D KS 6S159D 5S 2S 4C16""")17two_aces = board.parseDeck("""18KH 2H 4D QC 3S 7D 7S KD19JD 4H 3C 7C 9H KS 7H 5D206S AC QS TD JC 9C JH KC219S 4S QH 2D 4C AD TS 8H22TH 2S JS 6C 9D 3D 8D TC232C 5H 8C 5S 3H 6H 8S 6D24AH 5C QD AS25""")26two_aces_two = board.parseDeck("""276H AH 3H 2D JC 4D TH QD28TS 7D 9D 2C 7S QS JD 5D298H 4S 6S 5H 6C KD KC JH30QC TD 8S 7C 4C KS 9H 3S312S 2H 9S TC 8D 5C AD 4H32AS 5S 8C QH JS 6D KH 7H33AC 3D 9C 3C34""")35class CardUnitTest(unittest.TestCase):36 def test_makeCard(self):37 for card in range(0,52):38 self.assertEqual(card, board.makeCard(card // 13, card % 13))39 def test_suit(self):40 for card in range(0,52):41 self.assertEqual(card // 13, board.suit(card), )42 def test_pips(self):43 for card in range(0,52):44 self.assertEqual(card % 13, board.pips(card))45 def assert_formatCard(self, formatted, suit, pips):46 card = board.makeCard(suit, pips)47 actual = board.formatCard(card)48 self.assertEqual(formatted, actual, f"Card format failed %{pips}; {suit}")49 actual = board.parseCard(formatted)50 self.assertEqual(card, actual, f"Card parse %{pips}; {suit}")51 def test_formatCard(self):52 self.assert_formatCard('AC', 0, 0)53 self.assert_formatCard('2C', 0, 1)54 self.assert_formatCard('9C', 0, 8)55 self.assert_formatCard('TC', 0, 9)56 self.assert_formatCard('JC', 0, 10)57 self.assert_formatCard('QC', 0, 11)58 self.assert_formatCard('KC', 0, 12)59class BoardUnitTest(unittest.TestCase):60 def assert_init(self, deck):61 cards = len(deck)62 self.assertEqual(0, cards % 13, f"Uneven suits: {cards} cards in the deck")63 b = board.Board(deck)64 suits = cards // 1365 self.assertEqual(suits, b._nsuits)66 self.assertEqual(suits, len(b._foundations))67 for a, ace in enumerate(b._foundations):68 self.assertEqual(ace, board.noCard, f"Ace {a} not empty")69 self.assertEqual(suits * 2, len(b._tableau))70 for c, cascade in enumerate(b._tableau):71 expected = cards // len(b._tableau) + (c < suits)72 self.assertEqual(expected, len(cascade), f"Column {c} incorrectly sized")73 for r, card in enumerate(cascade):74 expected = deck[r * suits * 2 + c]75 self.assertEqual(expected, card)76 self.assertIsNone(b._memento, "Pre-initialised memento")77 self.assertTrue(b._resort, "Not initialised to resort")78 self.assertTrue(b._rehash, "Not initialised to rehash")79 self.assertEqual(len(b._tableau), len(b._sorted), "Hashing cascades not initialised")80 for s, actual in enumerate(b._sorted):81 self.assertEqual(b._tableau[s], actual)82 def test_init(self):83 self.assert_init(unshuffled)84 def assert_str(self, expected, deck):85 self.assertEqual(expected, str(board.Board(deck)))86 def test_str_unshuffled(self):87 expected = """Aces:88-S -H -D -C89Table:90AC 2C 3C 4C 5C 6C 7C 8C919C TC JC QC KC AD 2D 3D924D 5D 6D 7D 8D 9D TD JD93QD KD AH 2H 3H 4H 5H 6H947H 8H 9H TH JH QH KH AS952S 3S 4S 5S 6S 7S 8S 9S96TS JS QS KS -- -- -- --97Cells:98-- -- -- --99"""100 self.assert_str(expected, unshuffled)101 def test_str_reversed(self):102 expected = """Aces:103-S -H -D -C104Table:105KS QS JS TS 9S 8S 7S 6S1065S 4S 3S 2S AS KH QH JH107TH 9H 8H 7H 6H 5H 4H 3H1082H AH KD QD JD TD 9D 8D1097D 6D 5D 4D 3D 2D AD KC110QC JC TC 9C 8C 7C 6C 5C1114C 3C 2C AC -- -- -- --112Cells:113-- -- -- --114"""115 self.assert_str(expected, reversed)116 def test_str_two_aces(self):117 expected = """Aces:118-S -H -D -C119Table:120KH 2H 4D QC 3S 7D 7S KD121JD 4H 3C 7C 9H KS 7H 5D1226S AC QS TD JC 9C JH KC1239S 4S QH 2D 4C AD TS 8H124TH 2S JS 6C 9D 3D 8D TC1252C 5H 8C 5S 3H 6H 8S 6D126AH 5C QD AS -- -- -- --127Cells:128-- -- -- --129"""130 self.assert_str(expected, two_aces)131 def test_move_to_foundations_no_aces(self):132 b = board.Board(no_aces)133 b._rehash = b._resort = False134 actual = b.moveToFoundations()135 expected = []136 self.assertEqual(expected, actual)137 self.assertFalse(b._rehash)138 self.assertFalse(b._resort)139 def test_move_to_foundations_two_aces(self):140 b = board.Board(two_aces)141 b._rehash = b._resort = False142 actual = b.moveToFoundations()143 expected = [(0, -3,), (3, -4),]144 self.assertEqual(expected, actual)145 self.assertTrue(b._rehash)146 self.assertFalse(b._resort)147 def test_move_to_foundations_two_aces_two(self):148 b = board.Board(two_aces_two)149 b._rehash = b._resort = False150 actual = b.moveToFoundations()151 expected = [(0, -1,), (0, -4), (0, -4), ]152 self.assertEqual(expected, actual)153 self.assertTrue(b._rehash)154 self.assertFalse(b._resort)155 def test_move_to_foundations_reversed(self):156 setup = board.Board(reversed)157 setup._rehash = setup._resort = False158 actual = setup.moveToFoundations()159 self.assertTrue(setup._rehash)160 self.assertTrue(setup._resort)161 def test_move_to_foundations_cell(self):162 setup = board.Board(unshuffled)163 # One king in a cell164 for cascade in setup._tableau: cascade.clear()165 king = 12166 queen = king - 1167 clubs = 0168 setup._cells = [ board.noCard, board.makeCard(clubs, king), board.noCard, board.noCard, ]169 setup._foundations = [queen, king, king, king, ]170 expected = [(9, -1,),]171 actual = setup.moveToFoundations()172 self.assertEqual(expected, actual)173 def test_move_between_cascades_and_cells(self):174 b = board.Board(unshuffled)175 # Move to cell #1176 b.moveCard((0,8,))177 self.assertTrue(b._rehash)178 self.assertEqual(1, b._firstFree)179 self.assertEqual(6, len(b._tableau[0]))180 # Move to cell #2181 b.moveCard((0,9,))182 self.assertTrue(b._rehash)183 self.assertEqual(2, b._firstFree)184 self.assertEqual(5, len(b._tableau[0]))185 # Move to cell #3186 b.moveCard((1,10,))187 self.assertTrue(b._rehash)188 self.assertEqual(3, b._firstFree)189 self.assertEqual(6, len(b._tableau[1]))190 # Move to cell #4191 b.moveCard((2,11,))192 self.assertTrue(b._rehash)193 self.assertEqual(4, b._firstFree)194 self.assertEqual(6, len(b._tableau[2]))195 # Move to cascade #1196 b.moveCard((11,3,))197 self.assertTrue(b._rehash)198 self.assertEqual(3, b._firstFree)199 self.assertEqual(8, len(b._tableau[3]))200 # Move to cascade #2201 b.moveCard((10,3,))202 self.assertTrue(b._rehash)203 self.assertEqual(2, b._firstFree)204 self.assertEqual(9, len(b._tableau[3]))205 # Move to cascade #3206 b.moveCard((8,3,))207 self.assertTrue(b._rehash)208 self.assertEqual(0, b._firstFree)209 self.assertEqual(10, len(b._tableau[3]))210 # Move to cascade #4211 b.moveCard((9,1,))212 self.assertTrue(b._rehash)213 self.assertEqual(0, b._firstFree)214 self.assertEqual(7, len(b._tableau[1]))215 def test_move_between_cascades_and_foundations(self):216 b = board.Board(two_aces_two)217 # Move to foundations #1218 b._resort = b._rehash = False219 b.moveCard((0,-1,))220 self.assertTrue(b._rehash)221 self.assertFalse(b._resort)222 self.assertEqual(0, b._firstFree)223 self.assertEqual(6, len(b._tableau[0]))224 self.assertEqual([0, board.noCard, board.noCard, board.noCard,], b._foundations)225 # Move to foundations #2226 b._resort = b._rehash = False227 b.moveCard((0,-4,))228 self.assertTrue(b._rehash)229 self.assertFalse(b._resort)230 self.assertEqual(0, b._firstFree)231 self.assertEqual(5, len(b._tableau[0]))232 self.assertEqual([0, board.noCard, board.noCard, 0,], b._foundations)233 # Move to foundations #3234 b._resort = b._rehash = False235 b.moveCard((0,-4,))236 self.assertTrue(b._rehash)237 self.assertFalse(b._resort)238 self.assertEqual(0, b._firstFree)239 self.assertEqual(4, len(b._tableau[0]))240 self.assertEqual([0, board.noCard, board.noCard, 1,], b._foundations)241 # Move from foundations #1242 b._resort = b._rehash = False243 b.moveCard((-4,0,), False)244 self.assertTrue(b._rehash)245 self.assertFalse(b._resort)246 self.assertEqual(0, b._firstFree)247 self.assertEqual(5, len(b._tableau[0]))248 self.assertEqual([0, board.noCard, board.noCard, 0,], b._foundations)249 # Move from foundations #2250 b._resort = b._rehash = False251 b.moveCard((-4,0,), False)252 self.assertTrue(b._rehash)253 self.assertFalse(b._resort)254 self.assertEqual(0, b._firstFree)255 self.assertEqual(6, len(b._tableau[0]))256 self.assertEqual([0, board.noCard, board.noCard, board.noCard,], b._foundations)257 # Move from foundations #2258 b._resort = b._rehash = False259 b.moveCard((-1,0,), False)260 self.assertTrue(b._rehash)261 self.assertFalse(b._resort)262 self.assertEqual(0, b._firstFree)263 self.assertEqual(7, len(b._tableau[0]))264 self.assertEqual([board.noCard, board.noCard, board.noCard, board.noCard,], b._foundations)265 def test_move_to_empty_cascade(self):266 b = board.Board(unshuffled)267 # Put kings on the foundations268 king = 12269 for f, foundation in enumerate(b._foundations): b._foundations[f] = king270 # Clear the cascades271 for cascade in b._tableau: cascade.clear()272 # Move a king to an empty cell273 b.moveCard((-1, 0,))274 self.assertEqual(king, b._tableau[0][-1])275 self.assertTrue(b._resort)276 self.assertTrue(b._rehash)277 # Move a queen to an empty cell278 queen = king - 1279 b._resort = b._rehash = False280 b.moveCard((-1, 1,))281 self.assertEqual(queen, b._tableau[1][-1])282 self.assertTrue(b._resort)283 self.assertTrue(b._rehash)284 def test_enumerate_finish_to_card(self):285 b = board.Board(no_aces)286 start = 1287 actual = b.enumerateFinishCascades(start, b._tableau[start][-1])288 expected = [(start, 7,),]289 self.assertEqual(expected, actual)290 def test_enumerate_finish_to_empty(self):291 setup = board.Board(unshuffled)292 # Set up an empty cascade293 setup._foundations = [12 - len(setup._tableau) + 1, 12, 12, 12,]294 # Clubs in the top row, except the first cascade295 for start, cascade in enumerate(setup._tableau):296 cascade.clear()297 if start: cascade.append(start)298 # Every card can move to the leftmost or the next cascade299 for start, cascade in enumerate(setup._tableau):300 if cascade:301 actual = setup.enumerateFinishCascades(start, cascade[-1])302 expected = []303 if start + 1 < len(setup._tableau):304 expected.append( (start, start + 1,) )305 self.assertEqual(expected, actual, f"From cascade {start}" )306 def test_enumerate_moves_unshuffled(self):307 setup = board.Board(unshuffled)308 width = len(setup._tableau)309 # 3. Move from cascades to the next open width310 expected = [(start, width,) for start in range(width)]311 # 2. Move from cascades to cascades312 for start in range(width):313 if start != 3:314 expected.append((start, (start + 1) % width,))315 # 1. Move from cells to cascades316 # ...317 actual = setup.enumerateMoves()318 self.assertEqual(expected, actual)319 def test_enumerate_moves_no_aces(self):320 setup = board.Board(no_aces)321 width = len(setup._tableau)322 # 3. Move from cascades to the next open cell323 expected = [(start, width,) for start in range(width)]324 # 2. Move from cascades to cascades325 expected.append( (1, 7,) )326 # 1. Move from cells to cascades327 actual = setup.enumerateMoves()328 self.assertEqual(expected, actual)329 def test_enumerate_moves_two_aces(self):330 setup = board.Board(two_aces)331 width = len(setup._tableau)332 # 3. Move from cascades to the next open cell333 expected = [(start, width,) for start in range(width)]334 # 2. Move from cascades to cascades335 # 1. Move from cells to cascades336 actual = setup.enumerateMoves()337 self.assertEqual(expected, actual)338 def test_enumerate_moves_two_aces_two(self):339 setup = board.Board(two_aces_two)340 width = len(setup._tableau)341 # Clear the aces342 setup.moveToFoundations()343 # First move options - cells only344 expected = [(start, width,) for start in range(width)]345 actual = setup.enumerateMoves()346 self.assertEqual( expected, actual )347 # Uncover QH348 setup.moveCard( (3, width,) )349 expected = [(start, width + 1,) for start in range(width)]350 expected.append( (3, 6,) )351 actual = setup.enumerateMoves()352 self.assertEqual(expected, actual)353 def test_enumerate_moves_keep_sequences(self):354 setup = board.Board(unshuffled)355 setup._foundations = [8, 6, 8, 7, ]356 cascades = ( "TC 9C", "7H 9S 9D KH QH JH TH", "QC", "QD JD TD", "8C KD", "9H 8H", "KS QS JS", "",)357 for t, s in enumerate( cascades ):358 setup._tableau[ t ] = board.parseDeck( s )359 setup._cells = board.parseDeck( "KC TS JC --" )360 setup._firstFree = 3361 # Validate stacking362 stacked = (0, 1, 3, 5, 6, )363 for c in stacked: self.assertTrue( board.isStacked( setup._tableau[ c ] ) )364 isolate = (2, 4, )365 for c in isolate: self.assertFalse( board.isStacked( setup._tableau[ c ] ) )366 opens = (7, )367 for c in opens: self.assertFalse( board.isStacked( setup._tableau[ c ] ) )368 expected = []369 # Move stacked cascades to open cell370 expected.extend( [ (start, 11, ) for start in stacked ] )371 expected.pop()372 # Move isolate cascades to open cell373 expected.extend( [ (start, 11, ) for start in isolate ] )374 # Move cell 1 to open cascade375 expected.append( ( 8, 7, ) )376 # Move cell 2 to stack377 expected.append( ( 9, 6, ) )378 # Move cell 2 to open cascade379 expected.append( ( 9, 7, ) )380 # Move cell 3 to stack381 expected.append( ( 10, 2, ) )382 # Move cell 3 to open cascade383 expected.append( ( 10, 7, ) )384 # Move stacked cascades to open cascade385 expected.extend( [ (start, 7, ) for start in stacked ] )386 expected.pop()387 # Move isolate cascades to open cascade388 expected.extend( [ (start, 7, ) for start in isolate ] )389 expected.pop( -2 )390 actual = setup.enumerateMoves()391 self.assertEqual(expected, actual)392 def test_enumerate_moves_cover_aces(self):393 setup = board.Board(unshuffled)394 setup._foundations = [4, -1, 12, 1, ]395 cascades = (396 "QS",397 "KD QD JD TD 9D 8D 7D 6D 5D 4D",398 "JC",399 "JS TS 9S 8S",400 "KC QC JC TC 9C 8C 7C",401 "2D",402 "3S AD KS",403 "7S 6S",)404 for t, s in enumerate( cascades ):405 setup._tableau[ t ] = board.parseDeck( s )406 setup._cells = board.parseDeck( "4S -- 5S 3D" )407 setup._firstFree = 1408 expected = [(3, 9), (7, 9), (0, 9), (2, 9), (5, 9), (6, 9), (10, 7), (11, 1), (0, 6), ]409 actual = setup.enumerateMoves()410 self.assertEqual(expected, actual)411 def test_memento_values(self):412 visited = set()413 for deck in (unshuffled, reversed, no_aces, two_aces, two_aces_two,):414 setup = board.Board(deck)415 self.assertTrue(setup._resort)416 self.assertTrue(setup._rehash)417 memento = setup.memento()418 self.assertFalse(setup._resort)419 self.assertFalse(setup._rehash)420 self.assertEqual(memento, setup._memento)421 self.assertFalse( memento in visited, f"Duplicate memento #{len(visited)}: {memento}" )422 visited.add(memento)423 # Check that a new copy gets the same hash424 setup = board.Board(deck)425 self.assertEqual(memento, setup.memento())426 def test_not_solved(self):427 setup = board.Board(unshuffled)428 self.assertFalse( setup.solved() )429 def test_solved(self):430 setup = board.Board(reversed)431 setup.moveToFoundations()432 self.assertTrue( setup.solved() )433 def test_backtrack_foundations(self):434 setup = board.Board(reversed)435 expected = str(setup)436 # Backtracking should undo everything437 moves = setup.moveToFoundations()438 setup.backtrack( moves )439 actual = str(setup)440 self.assertEqual(expected, actual)441 def assert_backtrack(self, deck):442 setup = board.Board(deck)443 setup.moveToFoundations()444 moves = setup.enumerateMoves()445 for move in moves:446 expected = str(setup)447 setup.moveCard(move)448 setup.backtrack( [move,] )449 actual = str(setup)450 self.assertEqual(expected, actual, move)451 def test_backtrack_no_aces(self):452 self.assert_backtrack(no_aces)453 def test_backtrack_two_aces(self):454 self.assert_backtrack(two_aces)455 def test_backtrack_two_aces_two(self):456 self.assert_backtrack(two_aces_two)457 def assert_solve(self, setup, expected, display = False):458 b = board.Board(setup)459 solution = b.solve()460 actual = len(solution)461 self.assertEqual(expected, actual)462 if display:463 forward = []464 while solution:465 moves = solution.pop()466 forward.append(moves.copy())467 b.backtrack(moves)468 forward.reverse()469 for moves in forward:470 print(moves)471 for move in moves:472 b.moveCard(move, False)473 print(b)474 def test_solve_unshuffled(self):475 self.assert_solve(unshuffled, 958)476 def test_solve_reversed(self):477 self.assert_solve(reversed, 1)478 def test_solve_no_aces(self):479 self.assert_solve(no_aces, 555, False)480 def test_solve_two_aces(self):481 self.assert_solve(two_aces, 86, False)482 def test_solve_two_aces_two(self):483 self.assert_solve(two_aces_two, 72)484if __name__ == '__main__':...
borg.py
Source:borg.py
1import datetime2import dateutil.parser3import json4import os5import subprocess6from prometheus_client import Gauge7class FileBackup():8 def parse(self, dict):9 self._name = dict['name']10 self._time = dateutil.parser.isoparse(dict['time'])11 return self12 def getName(self):13 return self._name14 def inject(self, list):15 list.append(self)16 return self17 def setTimestamp(self, gauge):18 gauge.set(self._time.timestamp())19 return self20 def print(self):21 print('- '+self._name+' made at '+str(self._time) )22class Borg:23 def __init__(self):24 self._port = 2325 self._borgPassword = os.environ.get('BORG_PASSWORD')26 self._encryptionMode = 'repokey-blake2'27 def resort(self, resort):28 self._resort = resort29 return self30 def user(self, user):31 self._user = user32 return self33 def host(self, host):34 self._host = host35 return self36 def port(self, port):37 self._port = port38 return self39 def path(self, path):40 self._path = path41 return self42 def keyFilePath(self, keyFilePath):43 self._keyFilePath = keyFilePath44 return self45 def borgPassword(self, borgPassword):46 self._borgPassword = borgPassword47 return self48 def init(self, copies):49 repositoryNumber = 150 while repositoryNumber <= copies:51 self.__createRepository(repositoryNumber)52 repositoryNumber += 153 print("Created")54 repositoryNumber = 155 while repositoryNumber <= copies:56 self.__initRepository(repositoryNumber)57 repositoryNumber += 158 def __createRepository(self, repositoryNumber):59 try:60 fileRepository = 'repo'+str(repositoryNumber)61 self._resort.adapter('files').createFolder(fileRepository)62 print("Folder for Repository "+str(repositoryNumber)+" created")63 except OSError:64 print("Folder for Repository "+str(repositoryNumber)+" already exists")65 def __initRepository(self, repositoryNumber):66 repo = self.__makeRepo(repositoryNumber)67 print("Initializing Repository "+str(repositoryNumber))68 completedProcess = self.command([69 'init',70 '-e',71 self._encryptionMode,72 repo73 ], repositoryNumber)74 if completedProcess.returncode != 0:75 print("Process did not return success:")76 print("Code: "+ str(completedProcess.returncode))77 print( str(completedProcess.stdout) )78 print( str(completedProcess.stderr) )79 return self80 print("Initialized Repository "+str(repositoryNumber)+" successfully")81 return self82 def remove(self, name, repositoryNumber):83 self.__findBackup(name, repositoryNumber)84 print("Removing backup "+name+" from Repository "+str(repositoryNumber))85 completedProcess = self.command([86 'delete',87 '::'+name,88 ], repositoryNumber)89 if completedProcess.returncode != 0:90 print("Process did not return success:")91 print("Code: "+ str(completedProcess.returncode))92 print( completedProcess.stdout.decode('utf-8') )93 print( completedProcess.stderr.decode('utf-8') )94 return self95 print("Removal of "+name+" from Repository "+str(repositoryNumber)+" finished successfully")96 return self97 def backup(self, name, target, repositoryNumber):98 print("Backing up "+target+" to Repository "+str(repositoryNumber))99 completedProcess = self.command([100 'create',101 '::'+name,102 '.'103 ], repositoryNumber, directory=target, capture_output=False)104 print("Backup of "+target+" to Repository "+str(repositoryNumber)+" finished successfully")105 return self106 def umount(self, target):107 completedProcess = subprocess.run([108 'fusermount',109 '-u',110 target111 ],112 capture_output=True113 )114 if completedProcess.returncode != 0:115 print("Failed to unmount "+target)116 print("Code: "+ str(completedProcess.returncode))117 print( completedProcess.stdout.decode('utf-8') )118 print( completedProcess.stderr.decode('utf-8') )119 return self120 def mount(self, name, target, repositoryNumber):121 print("Mounting backup "+name+" from Repository "+str(repositoryNumber)+' to '+target)122 print("The borg mount is run in foreground to facilitate usage inside Docker")123 print("Please cancel the program with an interrupt (control+c) after you are done.")124 completedProcess = self.command([125 'mount',126 '-f',127 '::'+name,128 target129 ], repositoryNumber, check=False)130 if completedProcess.returncode != 0:131 print("Process did not return success:")132 print("Code: "+ str(completedProcess.returncode))133 print( completedProcess.stdout.decode('utf-8') )134 print( completedProcess.stderr.decode('utf-8') )135 return self136 print("Mounted backup "+name+" from Repository "+str(repositoryNumber)+' to '+target)137 return self138 def list(self, repositoryNumber):139 completedProcess = self.command([140 'list',141 '--json',142 '::'143 ], repositoryNumber, check=False)144 if completedProcess.returncode != 0:145 print("Process did not return success:")146 print("Code: "+ str(completedProcess.returncode))147 print( completedProcess.stdout.decode('utf-8') )148 print( completedProcess.stderr.decode('utf-8') )149 raise ValueError("List process failed")150 output = completedProcess.stdout.decode('utf-8')151 decodedOutput = json.loads(output)152 return self.__archivesToBackups(decodedOutput['archives'])153 def __archivesToBackups(self, list):154 backups = []155 for archive in list:156 backup = FileBackup().parse(archive)157 backup.inject(backups)158 sortedBackups = sorted(backups, key=lambda backup : backup._time)159 return sortedBackups160 161 162 def restore(self, name, target, repositoryNumber):163 backup = self.__findBackup(name, repositoryNumber)164 print("Restoring backup "+backup.getName()+" from Repository "+str(repositoryNumber)+' to '+target)165 completedProcess = self.command([166 'extract',167 '::'+backup.getName()168 ], repositoryNumber, directory=target)169 def prune(self, repositoryNumber):170 repo = self.__makeRepo(repositoryNumber)171 print("Pruning file backups in repository "+repositoryNumber)172 completedProcess = self.command([173 'prune',174 '--keep-daily=14',175 '--keep-weekly=6',176 '--keep-monthly=6',177 '::'178 ], repositoryNumber)179 if completedProcess.returncode != 0:180 print("Process did not return success:")181 print("Code: "+ str(completedProcess.returncode))182 print( completedProcess.stdout.decode('utf-8') )183 print( completedProcess.stderr.decode('utf-8') )184 return self185 return completedProcess.stdout.decode('utf-8')186 def command(self, args, repoNumber, directory=None, check=True,187 capture_output=True):188 if directory is None:189 directory = os.getcwd()190 return subprocess.run(191 ['borgbackup'] + args,192 capture_output=capture_output,193 env={194 'BORG_NEW_PASSPHRASE': self._borgPassword,195 'BORG_PASSPHRASE': self._borgPassword,196 'BORG_REPO': self.__makeRepo(repoNumber),197 'SSH_AUTH_SOCK': os.environ.get('SSH_AUTH_SOCK', ''),198 'BORG_RSH': "ssh -o StrictHostKeyChecking=accept-new -i "+self._keyFilePath199 }, cwd=directory, check=check)200 def __makeRepo(self, number):201 return 'ssh://'+self._user+'@'+self._host+':'+str(self._port)+'/.'+self._path+'/repo'+str(number)202 def __findBackup(self, target, repositoryNumber):203 if target == 'latest':204 return self.list(repositoryNumber)[-1]205 206 return target207 def getRepositories(self):208 repos = []209 for directoryName in self._resort.adapter('files').listFolders():210 start = len('repo')211 end = len(directoryName)212 repos.append( directoryName[start:end] )213 return repos214 215 def scrape(self, gauge):216 for repositoryNumber in self.getRepositories():217 try:218 backup = self.__findBackup('latest', repositoryNumber)219 except IndexError:220 gauge.labels(self._resort._name, 'repository_'+str(repositoryNumber)).set(0)221 continue222 backup.setTimestamp( gauge.labels(self._resort._name, 'repository_'+str(repositoryNumber)) )...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run autotest automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
