pg_probackup

1
import os
2
import unittest
3
import random
4
import shutil
5

6
from .helpers.cfs_helpers import find_by_extensions, find_by_name, find_by_pattern, corrupt_file
7
from .helpers.ptrack_helpers import ProbackupTest, ProbackupException
8

9

10
class CfsCatchupNoEncTest(ProbackupTest, unittest.TestCase):
11

12
    @unittest.skipUnless(ProbackupTest.enterprise, 'skip')
13
    def test_full_catchup_with_tablespace(self):
14
        """
15
        Test tablespace transfers
16
        """
17
        # preparation
18
        src_pg = self.make_simple_node(
19
            base_dir = os.path.join(self.module_name, self.fname, 'src'),
20
            set_replication = True
21
        )
22
        src_pg.slow_start()
23
        tblspace1_old_path = self.get_tblspace_path(src_pg, 'tblspace1_old')
24
        self.create_tblspace_in_node(src_pg, 'tblspace1', tblspc_path = tblspace1_old_path, cfs=True)
25
        src_pg.safe_psql(
26
            "postgres",
27
            "CREATE TABLE ultimate_question TABLESPACE tblspace1 AS SELECT 42 AS answer")
28
        src_query_result = src_pg.table_checksum("ultimate_question")
29
        src_pg.safe_psql(
30
            "postgres",
31
            "CHECKPOINT")
32

33
        # do full catchup with tablespace mapping
34
        dst_pg = self.make_empty_node(os.path.join(self.module_name, self.fname, 'dst'))
35
        tblspace1_new_path = self.get_tblspace_path(dst_pg, 'tblspace1_new')
36
        self.catchup_node(
37
            backup_mode = 'FULL',
38
            source_pgdata = src_pg.data_dir,
39
            destination_node = dst_pg,
40
            options = [
41
                '-d', 'postgres',
42
                '-p', str(src_pg.port),
43
                '--stream',
44
                '-T', '{0}={1}'.format(tblspace1_old_path, tblspace1_new_path)
45
            ]
46
        )
47

48
        # 1st check: compare data directories
49
        self.compare_pgdata(
50
            self.pgdata_content(src_pg.data_dir),
51
            self.pgdata_content(dst_pg.data_dir)
52
        )
53

54
        # check cfm size
55
        cfms = find_by_extensions([os.path.join(dst_pg.data_dir)], ['.cfm'])
56
        self.assertTrue(cfms, "ERROR: .cfm files not found in backup dir")
57
        for cfm in cfms:
58
            size = os.stat(cfm).st_size
59
            self.assertLessEqual(size, 4096,
60
                                 "ERROR: {0} is not truncated (has size {1} > 4096)".format(
61
                                     cfm, size
62
                                 ))
63

64
        # make changes in master tablespace
65
        src_pg.safe_psql(
66
            "postgres",
67
            "UPDATE ultimate_question SET answer = -1")
68
        src_pg.safe_psql(
69
            "postgres",
70
            "CHECKPOINT")
71

72
        # run&recover catchup'ed instance
73
        dst_options = {}
74
        dst_options['port'] = str(dst_pg.port)
75
        self.set_auto_conf(dst_pg, dst_options)
76
        dst_pg.slow_start()
77

78
        # 2nd check: run verification query
79
        dst_query_result = dst_pg.table_checksum("ultimate_question")
80
        self.assertEqual(src_query_result, dst_query_result, 'Different answer from copy')
81

82
        # and now delta backup
83
        dst_pg.stop()
84

85
        self.catchup_node(
86
            backup_mode = 'DELTA',
87
            source_pgdata = src_pg.data_dir,
88
            destination_node = dst_pg,
89
            options = [
90
                '-d', 'postgres',
91
                '-p', str(src_pg.port),
92
                '--stream',
93
                '-T', '{0}={1}'.format(tblspace1_old_path, tblspace1_new_path)
94
            ]
95
        )
96

97
        # check cfm size again
98
        cfms = find_by_extensions([os.path.join(dst_pg.data_dir)], ['.cfm'])
99
        self.assertTrue(cfms, "ERROR: .cfm files not found in backup dir")
100
        for cfm in cfms:
101
            size = os.stat(cfm).st_size
102
            self.assertLessEqual(size, 4096,
103
                                 "ERROR: {0} is not truncated (has size {1} > 4096)".format(
104
                                     cfm, size
105
                                 ))
106

107
        # run&recover catchup'ed instance
108
        dst_options = {}
109
        dst_options['port'] = str(dst_pg.port)
110
        self.set_auto_conf(dst_pg, dst_options)
111
        dst_pg.slow_start()
112

113

114
        # 3rd check: run verification query
115
        src_query_result = src_pg.table_checksum("ultimate_question")
116
        dst_query_result = dst_pg.table_checksum("ultimate_question")
117
        self.assertEqual(src_query_result, dst_query_result, 'Different answer from copy')
118