db/pg_probackup
1
0
Fork 0
mirror of https://github.com/postgrespro/pg_probackup.git synced 2025-07-06 05:57:21 +02:00
Code Issues Releases Activity

ptrack tests added

Browse Source
This commit is contained in:
Grigory Smolkin
2017-05-03 14:14:48 +03:00
parent d4743168df
commit 2b8da45f09
21 changed files with 2918 additions and 1248 deletions
Download Patch File Download Diff File Expand all files Collapse all files

134
tests/delete_test.py
View File

@ -1,88 +1,102 @@
import unittest
from os import path
import six
from .pb_lib import ProbackupTest
from .ptrack_helpers import ProbackupTest, ProbackupException
from testgres import stop_all
import subprocess
class DeleteTest(ProbackupTest, unittest.TestCase):
def __init__(self, *args, **kwargs):
super(DeleteTest, self).__init__(*args, **kwargs)
def __init__(self, *args, **kwargs):
super(DeleteTest, self).__init__(*args, **kwargs)
@classmethod
def tearDownClass(cls):
stop_all()
# @classmethod
# def tearDownClass(cls):
# stop_all()
# @unittest.skip("123")
def test_delete_full_backups(self):
"""delete full backups"""
fname = self.id().split('.')[3]
print '{0} started'.format(fname)
node = self.make_simple_node(base_dir="tmp_dirs/delete/{0}".format(fname),
set_archiving=True,
initdb_params=['--data-checksums'],
pg_options={'wal_level': 'replica'}
)
node.start()
self.assertEqual(self.init_pb(node), six.b(""))
node.pgbench_init()
def test_delete_full_backups_1(self):
"""delete full backups"""
node = self.make_bnode(base_dir="tmp_dirs/delete/delete_full_backups_1")
node.start()
self.assertEqual(self.init_pb(node), six.b(""))
node.pgbench_init()
# full backup mode
with open(path.join(node.logs_dir, "backup_1.log"), "wb") as backup_log:
backup_log.write(self.backup_pb(node, options=["--verbose"]))
# full backup mode
with open(path.join(node.logs_dir, "backup_1.log"), "wb") as backup_log:
backup_log.write(self.backup_pb(node, options=["--verbose"]))
pgbench = node.pgbench(stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
pgbench.wait()
pgbench.stdout.close()
pgbench = node.pgbench(stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
pgbench.wait()
pgbench.stdout.close()
with open(path.join(node.logs_dir, "backup_2.log"), "wb") as backup_log:
backup_log.write(self.backup_pb(node, options=["--verbose"]))
with open(path.join(node.logs_dir, "backup_2.log"), "wb") as backup_log:
backup_log.write(self.backup_pb(node, options=["--verbose"]))
pgbench = node.pgbench(stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
pgbench.wait()
pgbench.stdout.close()
pgbench = node.pgbench(stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
pgbench.wait()
pgbench.stdout.close()
with open(path.join(node.logs_dir, "backup_3.log"), "wb") as backup_log:
backup_log.write(self.backup_pb(node, options=["--verbose"]))
with open(path.join(node.logs_dir, "backup_3.log"), "wb") as backup_log:
backup_log.write(self.backup_pb(node, options=["--verbose"]))
show_backups = self.show_pb(node)
id_1 = show_backups[0]['ID']
id_3 = show_backups[2]['ID']
self.delete_pb(node, show_backups[1]['ID'])
show_backups = self.show_pb(node)
self.assertEqual(show_backups[0]['ID'], id_1)
self.assertEqual(show_backups[1]['ID'], id_3)
show_backups = self.show_pb(node)
id_1 = show_backups[0].id
id_2 = show_backups[2].id
self.delete_pb(node, show_backups[1].id)
show_backups = self.show_pb(node)
self.assertEqual(show_backups[0].id, id_1)
self.assertEqual(show_backups[1].id, id_2)
node.stop()
node.stop()
# @unittest.skip("123")
def test_delete_increment(self):
"""delete increment and all after him"""
fname = self.id().split('.')[3]
print '{0} started'.format(fname)
node = self.make_simple_node(base_dir="tmp_dirs/delete/{0}".format(fname),
set_archiving=True,
initdb_params=['--data-checksums'],
pg_options={'wal_level': 'replica'}
)
node.start()
self.assertEqual(self.init_pb(node), six.b(""))
def test_delete_increment_2(self):
"""delete increment and all after him"""
node = self.make_bnode(base_dir="tmp_dirs/delete/delete_increment_2")
node.start()
self.assertEqual(self.init_pb(node), six.b(""))
# full backup mode
with open(path.join(node.logs_dir, "backup_1.log"), "wb") as backup_log:
backup_log.write(self.backup_pb(node, options=["--verbose"]))
# full backup mode
with open(path.join(node.logs_dir, "backup_1.log"), "wb") as backup_log:
backup_log.write(self.backup_pb(node, options=["--verbose"]))
# page backup mode
with open(path.join(node.logs_dir, "backup_2.log"), "wb") as backup_log:
backup_log.write(self.backup_pb(node, backup_type="page", options=["--verbose"]))
# page backup mode
with open(path.join(node.logs_dir, "backup_2.log"), "wb") as backup_log:
backup_log.write(self.backup_pb(node, backup_type="page", options=["--verbose"]))
# page backup mode
with open(path.join(node.logs_dir, "backup_3.log"), "wb") as backup_log:
backup_log.write(self.backup_pb(node, backup_type="page", options=["--verbose"]))
# page backup mode
with open(path.join(node.logs_dir, "backup_3.log"), "wb") as backup_log:
backup_log.write(self.backup_pb(node, backup_type="page", options=["--verbose"]))
# full backup mode
self.backup_pb(node)
# full backup mode
self.backup_pb(node)
show_backups = self.show_pb(node)
show_backups = self.show_pb(node)
self.assertEqual(len(show_backups), 4)
self.assertEqual(len(show_backups), 4)
# delete first page backup
self.delete_pb(node, show_backups[2].id)
# delete first page backup
self.delete_pb(node, show_backups[1]['ID'])
show_backups = self.show_pb(node)
self.assertEqual(len(show_backups), 2)
show_backups = self.show_pb(node)
self.assertEqual(len(show_backups), 2)
self.assertEqual(show_backups[0].mode, six.b("FULL"))
self.assertEqual(show_backups[0].status, six.b("OK"))
self.assertEqual(show_backups[1].mode, six.b("FULL"))
self.assertEqual(show_backups[1].status, six.b("OK"))
self.assertEqual(show_backups[0]['Mode'], six.b("FULL"))
self.assertEqual(show_backups[0]['Status'], six.b("OK"))
self.assertEqual(show_backups[1]['Mode'], six.b("FULL"))
self.assertEqual(show_backups[1]['Status'], six.b("OK"))
node.stop()
node.stop()