86 'managed attribute. Is FSPATH() argument BFSS managed?') |
87 'managed attribute. Is FSPATH() argument BFSS managed?') |
87 |
88 |
88 def test_bfss_storage(self): |
89 def test_bfss_storage(self): |
89 with self.admin_access.repo_cnx() as cnx: |
90 with self.admin_access.repo_cnx() as cnx: |
90 f1 = self.create_file(cnx) |
91 f1 = self.create_file(cnx) |
91 expected_filepath = osp.join(self.tempdir, '%s_data_%s' % |
92 filepaths = glob(osp.join(self.tempdir, '%s_data_*' % f1.eid)) |
92 (f1.eid, f1.data_name)) |
93 self.assertEqual(len(filepaths), 1, filepaths) |
93 self.assertTrue(osp.isfile(expected_filepath)) |
94 expected_filepath = filepaths[0] |
94 # file should be read only |
95 # file should be read only |
95 self.assertFalse(os.access(expected_filepath, os.W_OK)) |
96 self.assertFalse(os.access(expected_filepath, os.W_OK)) |
96 self.assertEqual(file(expected_filepath).read(), 'the-data') |
97 self.assertEqual(file(expected_filepath).read(), 'the-data') |
97 cnx.rollback() |
98 cnx.rollback() |
98 self.assertFalse(osp.isfile(expected_filepath)) |
99 self.assertFalse(osp.isfile(expected_filepath)) |
99 f1 = self.create_file(cnx) |
100 filepaths = glob(osp.join(self.tempdir, '%s_data_*' % f1.eid)) |
100 cnx.commit() |
101 self.assertEqual(len(filepaths), 0, filepaths) |
|
102 f1 = self.create_file(cnx) |
|
103 cnx.commit() |
|
104 filepaths = glob(osp.join(self.tempdir, '%s_data_*' % f1.eid)) |
|
105 self.assertEqual(len(filepaths), 1, filepaths) |
|
106 expected_filepath = filepaths[0] |
101 self.assertEqual(file(expected_filepath).read(), 'the-data') |
107 self.assertEqual(file(expected_filepath).read(), 'the-data') |
102 f1.cw_set(data=Binary('the new data')) |
108 f1.cw_set(data=Binary('the new data')) |
103 cnx.rollback() |
109 cnx.rollback() |
104 self.assertEqual(file(expected_filepath).read(), 'the-data') |
110 self.assertEqual(file(expected_filepath).read(), 'the-data') |
105 f1.cw_delete() |
111 f1.cw_delete() |
112 |
118 |
113 def test_bfss_sqlite_fspath(self): |
119 def test_bfss_sqlite_fspath(self): |
114 with self.admin_access.repo_cnx() as cnx: |
120 with self.admin_access.repo_cnx() as cnx: |
115 f1 = self.create_file(cnx) |
121 f1 = self.create_file(cnx) |
116 expected_filepath = osp.join(self.tempdir, '%s_data_%s' % (f1.eid, f1.data_name)) |
122 expected_filepath = osp.join(self.tempdir, '%s_data_%s' % (f1.eid, f1.data_name)) |
117 self.assertEqual(self.fspath(cnx, f1), expected_filepath) |
123 base, ext = osp.splitext(expected_filepath) |
|
124 self.assertTrue(self.fspath(cnx, f1).startswith(base)) |
|
125 self.assertTrue(self.fspath(cnx, f1).endswith(ext)) |
118 |
126 |
119 def test_bfss_fs_importing_doesnt_touch_path(self): |
127 def test_bfss_fs_importing_doesnt_touch_path(self): |
120 with self.admin_access.repo_cnx() as cnx: |
128 with self.admin_access.repo_cnx() as cnx: |
121 cnx.transaction_data['fs_importing'] = True |
129 cnx.transaction_data['fs_importing'] = True |
122 filepath = osp.abspath(__file__) |
130 filepath = osp.abspath(__file__) |