Cyclone-Team/gramps
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Replace TEMP_DIR by tempfile object

Browse Source
This commit is contained in:
André Apitzsch 2020-11-18 10:20:57 +01:00 committed by Nick Hall
parent 1dd36390ff
commit fd26b5f066
8 changed files with 140 additions and 186 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
gramps/cli/argparser.py
View File

@ -48,7 +48,7 @@ from glob import glob
#
#-------------------------------------------------------------------------
from gramps.gen.const import (LONGOPTS, SHORTOPTS, USER_PLUGINS, VERSION_DIR,
HOME_DIR, TEMP_DIR, THUMB_DIR, ENV_DIR, USER_CSS)
HOME_DIR, THUMB_DIR, ENV_DIR, USER_CSS)
from gramps.gen.utils.cast import get_type_converter
from gramps.gen.const import GRAMPS_LOCALE as glocale
_ = glocale.translation.gettext
@ -423,7 +423,6 @@ class ArgParser:
for fil in glob(os.path.join(HOME_DIR, "*.zip")):
os.remove(fil)
if 'E' in value: # Everything else
rmtree(TEMP_DIR)
rmtree(THUMB_DIR)
rmtree(USER_CSS)
rmtree(ENV_DIR)

29
gramps/cli/test/cli_test.py
View File

@ -26,7 +26,6 @@ import unittest
import re
import subprocess
from gramps.gen.const import TEMP_DIR
from gramps.gen.dbstate import DbState
from gramps.test.test_util import Gramps
@ -113,34 +112,6 @@ class Test(unittest.TestCase):
g = re.search("INDI", content)
self.assertTrue(g, "found 'INDI' in output file")
# this verifies that files in the temporary "import dir"
# get cleaned before (and after) running a CLI
# (eg cleanout stale files from prior crash-runs)
def test3_files_in_import_dir(self):
ddir = os.path.join(TEMP_DIR, "import_dbdir")
try:
os.makedirs(ddir)
except:
pass
bogofiles = [os.path.join(ddir, fn) for fn in ("family.db", "lock")]
for fn in bogofiles:
with open(fn, "w") as f:
f.write("garbage")
# ~same as test 2
ifile = min1r
ofile = out_ged
gcmd = [sys.executable, "Gramps.py", "-i", ifile, "-e", ofile]
process = subprocess.Popen(gcmd,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
result_str, err_str = process.communicate()
self.assertEqual(process.returncode, 0,
"executed CLI command %r" % gcmd)
for fn in bogofiles:
self.assertFalse(os.path.exists(fn))
class UnicodeTest(unittest.TestCase):

4
gramps/gen/const.py
View File

@ -122,14 +122,13 @@ TOOL_OPTIONS = os.path.join(HOME_DIR, "tool_options.xml")
PLACE_FORMATS = os.path.join(HOME_DIR, "place_formats.xml")
ENV_DIR = os.path.join(HOME_DIR, "env")
TEMP_DIR = os.path.join(HOME_DIR, "temp")
THUMB_DIR = os.path.join(HOME_DIR, "thumb")
THUMB_NORMAL = os.path.join(THUMB_DIR, "normal")
THUMB_LARGE = os.path.join(THUMB_DIR, "large")
USER_PLUGINS = os.path.join(VERSION_DIR, "plugins")
USER_CSS = os.path.join(HOME_DIR, "css")
# dirs checked/made for each Gramps session
USER_DIRLIST = (USER_HOME, HOME_DIR, VERSION_DIR, ENV_DIR, TEMP_DIR, THUMB_DIR,
USER_DIRLIST = (USER_HOME, HOME_DIR, VERSION_DIR, ENV_DIR, THUMB_DIR,
THUMB_NORMAL, THUMB_LARGE, USER_PLUGINS, USER_CSS)
@ -199,7 +198,6 @@ ENV = {
"major_version": major_version,
"VERSION_DIR": VERSION_DIR,
"ENV_DIR": ENV_DIR,
"TEMP_DIR": TEMP_DIR,
"THUMB_DIR": THUMB_DIR,
"THUMB_NORMAL": THUMB_NORMAL,
"THUMB_LARGE": THUMB_LARGE,

100
gramps/gen/merge/test/merge_ref_test.py
View File

@ -29,11 +29,12 @@ from io import BytesIO
import difflib
import copy
import lxml.etree as ET
import tempfile
from gramps.plugins.lib.libgrampsxml import GRAMPS_XML_VERSION
from gramps.test.test_util import Gramps
from gramps.gen.user import User
from gramps.gen.const import DATA_DIR, USER_PLUGINS, TEMP_DIR
from gramps.gen.const import DATA_DIR, USER_PLUGINS
from gramps.version import VERSION
from gramps.gen.lib import Name, Surname
from gramps.gen.const import GRAMPS_LOCALE as glocale
@ -104,53 +105,43 @@ class BaseMergeCheck(unittest.TestCase):
def check_results(self, input_doc, expect_doc, result_str, err_str,
test_error_str=''):
input_file = os.path.join(TEMP_DIR, "merge_test_input.gramps")
try:
os.remove(input_file)
except OSError:
pass
with tempfile.TemporaryDirectory() as tmpdirname:
input_file = os.path.join(tmpdirname, "merge_test_input.gramps")
if err_str:
if test_error_str:
self.assertIn(test_error_str, err_str)
return
else:
if "Traceback (most recent call last):" in err_str:
inp = self.canonicalize(input_doc)
inpt = open(input_file, mode='wb')
inpt.write(inp)
inpt.close()
raise Exception(err_str)
result = self.canonicalize(result_str)
result_file = os.path.join(TEMP_DIR, "merge_test_result.gramps")
try:
os.remove(result_file)
except OSError:
pass
expect = self.canonicalize(expect_doc)
expect_file = os.path.join(TEMP_DIR, "merge_test_expected.gramps")
try:
os.remove(expect_file)
except OSError:
pass
if result != expect:
res = open(result_file, mode='wb')
res.write(result)
res.close()
eres = open(expect_file, mode='wb')
eres.write(expect)
eres.close()
inp = self.canonicalize(input_doc)
inpt = open(input_file, mode='wb')
inpt.write(inp)
inpt.close()
result = result.decode('utf-8')
expect = expect.decode('utf-8')
diff = difflib.ndiff(result, expect)
msg = ""
for line in diff:
msg += line
self.fail(msg)
if err_str:
if test_error_str:
self.assertIn(test_error_str, err_str)
return
else:
if "Traceback (most recent call last):" in err_str:
inp = self.canonicalize(input_doc)
inpt = open(input_file, mode='wb')
inpt.write(inp)
inpt.close()
raise Exception(err_str)
result = self.canonicalize(result_str)
result_file = os.path.join(tmpdirname, "merge_test_result.gramps")
expect = self.canonicalize(expect_doc)
expect_file = os.path.join(tmpdirname, "merge_test_expected.gramps")
if result != expect:
res = open(result_file, mode='wb')
res.write(result)
res.close()
eres = open(expect_file, mode='wb')
eres.write(expect)
eres.close()
inp = self.canonicalize(input_doc)
inpt = open(input_file, mode='wb')
inpt.write(inp)
inpt.close()
result = result.decode('utf-8')
expect = expect.decode('utf-8')
diff = difflib.ndiff(result, expect)
msg = ""
for line in diff:
msg += line
self.fail(msg)
def do_family_case(self, phoenix_id, titanic_id, father_h, mother_h,
input_doc, expect_doc, test_error_str=''):
@ -180,15 +171,12 @@ class BaseMergeCheck(unittest.TestCase):
msg = '\n***** result:\n' + result_str + \
'\n***** expect:\n' + expect_str
inp = self.canonicalize(input_doc)
input_file = os.path.join(TEMP_DIR, "merge_test_input.gramps")
try:
os.remove(input_file)
except OSError:
pass
inpt = open(input_file, mode='wb')
inpt.write(inp)
inpt.close()
self.fail(msg)
with tempfile.TemporaryDirectory() as tmpdirname:
input_file = os.path.join(tmpdirname, "merge_test_input.gramps")
inpt = open(input_file, mode='wb')
inpt.write(inp)
inpt.close()
self.fail(msg)
#-------------------------------------------------------------------------

10
gramps/gen/utils/file.py
View File

@ -32,6 +32,7 @@ File and folder related utility functions
import os
import sys
import shutil
import tempfile
import hashlib
import logging
LOG = logging.getLogger(".gen.utils.file")
@ -42,7 +43,7 @@ LOG = logging.getLogger(".gen.utils.file")
#
#-------------------------------------------------------------------------
from ..constfunc import win, mac, get_env_var
from ..const import TEMP_DIR, USER_HOME, ENV, GRAMPS_LOCALE as glocale
from ..const import USER_HOME, ENV, GRAMPS_LOCALE as glocale
#-------------------------------------------------------------------------
#
@ -94,15 +95,12 @@ def get_empty_tempdir(dirname):
or for inadequate permissions to delete dir/files or create dir(s)
"""
dirpath = os.path.join(TEMP_DIR, str(dirname))
if os.path.isdir(dirpath):
shutil.rmtree(dirpath)
os.makedirs(dirpath)
dirpath = tempfile.mkdtemp(prefix='gramps-', suffix='-'+dirname)
return dirpath
def rm_tempdir(path):
"""Remove a tempdir created with get_empty_tempdir"""
if path.startswith(TEMP_DIR) and os.path.isdir(path):
if path.startswith(tempfile.gettempdir()) and os.path.isdir(path):
shutil.rmtree(path)
def relative_path(original, base):

90
gramps/gen/utils/test/file_test.py
View File

@ -31,6 +31,7 @@ File tests.
#
#-------------------------------------------------------------------------
import os
import tempfile
import unittest
#-------------------------------------------------------------------------
@ -38,8 +39,8 @@ import unittest
# Gramps modules
#
#-------------------------------------------------------------------------
from ...const import TEMP_DIR, USER_HOME, USER_PLUGINS, VERSION
from ...utils.file import media_path, get_empty_tempdir
from ...const import USER_HOME, USER_PLUGINS, VERSION
from ...utils.file import media_path
from ...db.utils import make_database
#-------------------------------------------------------------------------
@ -56,55 +57,56 @@ class FileTest(unittest.TestCase):
"""
Test media path variables.
"""
with tempfile.TemporaryDirectory() as tmpdirname:
# Create database
db = make_database("sqlite")
path = os.path.join(tmpdirname, "utils_file_test")
os.makedirs(path)
db.load(path)
# Create database
db = make_database("sqlite")
path = get_empty_tempdir("utils_file_test")
db.load(path)
# Test without db.mediapath set
self.assertEqual(media_path(db), os.path.normcase(os.path.normpath(
os.path.abspath(USER_HOME))))
self.assertTrue(os.path.exists(media_path(db)))
# Test without db.mediapath set
self.assertEqual(media_path(db), os.path.normcase(os.path.normpath(
os.path.abspath(USER_HOME))))
self.assertTrue(os.path.exists(media_path(db)))
# Test with absolute db.mediapath
db.set_mediapath(os.path.abspath(USER_HOME) + "/test_abs")
self.assertEqual(media_path(db), os.path.normcase(os.path.normpath(
os.path.abspath(USER_HOME + "/test_abs"))))
# Test with absolute db.mediapath
db.set_mediapath(os.path.abspath(USER_HOME) + "/test_abs")
self.assertEqual(media_path(db), os.path.normcase(os.path.normpath(
os.path.abspath(USER_HOME + "/test_abs"))))
# Test with relative db.mediapath
db.set_mediapath("test_rel")
self.assertEqual(media_path(db), os.path.normcase(os.path.normpath(
os.path.abspath(tmpdirname + "/utils_file_test/test_rel"))))
# Test with relative db.mediapath
db.set_mediapath("test_rel")
self.assertEqual(media_path(db), os.path.normcase(os.path.normpath(
os.path.abspath(TEMP_DIR + "/utils_file_test/test_rel"))))
# Test with environment variable
db.set_mediapath("/test/{VERSION}/test_var")
self.assertEqual(media_path(db), os.path.normcase(os.path.normpath(
os.path.abspath("/test/" + VERSION + "/test_var"))))
db.set_mediapath("{USER_PLUGINS}/test_var")
self.assertEqual(media_path(db), os.path.normcase(os.path.normpath(
os.path.abspath(USER_PLUGINS + "/test_var"))))
db.set_mediapath("{VERSION}/test_var")
self.assertEqual(media_path(db), os.path.normcase(os.path.normpath(
os.path.abspath(tmpdirname + "/utils_file_test/" + VERSION +
"/test_var"))))
# Test with environment variable
db.set_mediapath("/test/{VERSION}/test_var")
self.assertEqual(media_path(db), os.path.normcase(os.path.normpath(
os.path.abspath("/test/" + VERSION + "/test_var"))))
db.set_mediapath("{USER_PLUGINS}/test_var")
self.assertEqual(media_path(db), os.path.normcase(os.path.normpath(
os.path.abspath(USER_PLUGINS + "/test_var"))))
db.set_mediapath("{VERSION}/test_var")
self.assertEqual(media_path(db), os.path.normcase(os.path.normpath(
os.path.abspath(TEMP_DIR + "/utils_file_test/" + VERSION +
"/test_var"))))
# Test with $GRAMPSHOME environment variable not set
old_env = os.environ.copy()
if 'GRAMPSHOME' in os.environ:
del os.environ['GRAMPSHOME']
db.set_mediapath("{GRAMPSHOME}/test_var")
self.assertEqual(media_path(db), os.path.normcase(os.path.normpath(
os.path.abspath(USER_HOME + "/test_var"))))
# Test with $GRAMPSHOME environment variable not set
old_env = os.environ.copy()
if 'GRAMPSHOME' in os.environ:
del os.environ['GRAMPSHOME']
db.set_mediapath("{GRAMPSHOME}/test_var")
self.assertEqual(media_path(db), os.path.normcase(os.path.normpath(
os.path.abspath(USER_HOME + "/test_var"))))
# Test with $GRAMPSHOME environment variable set
os.environ['GRAMPSHOME'] = "/this/is/a/test"
db.set_mediapath("{GRAMPSHOME}/test_var")
self.assertEqual(media_path(db), os.path.normcase(os.path.normpath(
os.path.abspath("/this/is/a/test/test_var"))))
# Test with $GRAMPSHOME environment variable set
os.environ['GRAMPSHOME'] = "/this/is/a/test"
db.set_mediapath("{GRAMPSHOME}/test_var")
self.assertEqual(media_path(db), os.path.normcase(os.path.normpath(
os.path.abspath("/this/is/a/test/test_var"))))
# Restore environment
os.environ = old_env
# Restore environment
os.environ = old_env
#-------------------------------------------------------------------------

36
gramps/plugins/test/exports_test.py
View File

@ -24,9 +24,10 @@ import os
import difflib
from unittest.mock import patch
from time import localtime, strptime
import tempfile
from gramps.test.test_util import Gramps
from gramps.gen.const import TEMP_DIR, DATA_DIR
from gramps.gen.const import DATA_DIR
from gramps.gen.datehandler import set_format
from gramps.gen.user import User
from gramps.gen.utils.config import config
@ -57,22 +58,23 @@ def do_it(srcfile, tstfile, dfilter=None):
"""
tst_file = os.path.join(TEST_DIR, srcfile)
expect_file = os.path.join(TEST_DIR, tstfile)
result_file = os.path.join(TEMP_DIR, tstfile)
err = call("-C", TREE_NAME, "-q",
"--import", tst_file,
"--export", result_file)[1]
if "Cleaning up." not in err:
return "Export failed, no 'Cleaning up.'"
msg = compare(expect_file, result_file, dfilter)
if not msg:
# we will leave the result_file in place if there was an error.
try:
os.remove(result_file)
except OSError:
pass
return
else:
return msg
with tempfile.TemporaryDirectory() as tmpdirname:
result_file = os.path.join(tmpdirname, tstfile)
err = call("-C", TREE_NAME, "-q",
"--import", tst_file,
"--export", result_file)[1]
if "Cleaning up." not in err:
return "Export failed, no 'Cleaning up.'"
msg = compare(expect_file, result_file, dfilter)
if not msg:
# we will leave the result_file in place if there was an error.
try:
os.remove(result_file)
except OSError:
pass
return
else:
return msg
def compare(expect_file, result_file, dfilter=None):

54
gramps/plugins/test/imports_test.py
View File

@ -37,7 +37,7 @@ from gramps.gen.merge.diff import diff_dbs, to_struct
from gramps.gen.simple import SimpleAccess
from gramps.gen.utils.id import set_det_id
from gramps.gen.user import User
from gramps.gen.const import TEMP_DIR, DATA_DIR
from gramps.gen.const import DATA_DIR
from gramps.test.test_util import capture
from gramps.plugins.export.exportxml import XmlWriter
@ -212,8 +212,6 @@ def make_tst_function(tstfile, file_name):
mockdtime.side_effect = mock_time
fn1 = os.path.join(TEST_DIR, tstfile)
fn2 = os.path.join(TEST_DIR, (file_name + ".gramps"))
fres = os.path.join(TEMP_DIR, (file_name + ".difs"))
fout = os.path.join(TEMP_DIR, (file_name + ".gramps"))
if "_dfs" in tstfile:
config.set('preferences.default-source', True)
config.set('preferences.tag-on-import-format', "Imported")
@ -223,11 +221,6 @@ def make_tst_function(tstfile, file_name):
skp_imp_adds = True
config.set('preferences.default-source', False)
config.set('preferences.tag-on-import', False)
try:
os.remove(fres)
os.remove(fout)
except OSError:
pass
#logger.info("\n**** %s ****", tstfile)
set_det_id(True)
with capture(None) as output:
@ -251,27 +244,30 @@ def make_tst_function(tstfile, file_name):
# Also we save the .gramps file from the import so user can perform
# Diff himself for better context.
if deltas:
writer = XmlWriter(self.database1, self.user,
strip_photos=0, compress=0)
writer.write(fout)
hres = open(fres, mode='w', encoding='utf-8',
errors='replace')
hres.write(self.msg)
hres.close()
# let's see if we have any allowed exception file
fdif = os.path.join(TEST_DIR, (file_name + ".difs"))
try:
hdif = open(fdif)
msg = hdif.read()
hdif.close()
except (FileNotFoundError, IOError):
msg = ""
# if exception file matches exactly, we are done.
if self.msg != msg:
self.msg = "\n****Captured Output****\n" + output[0] + \
"\n****Captured Err****\n" + output[1] + \
"\n****End Capture Err****\n" + self.msg
self.fail(self.msg)
with tempfile.TemporaryDirectory() as tmpdirname:
fres = os.path.join(tmpdirname, (file_name + ".difs"))
fout = os.path.join(tmpdirname, (file_name + ".gramps"))
writer = XmlWriter(self.database1, self.user,
strip_photos=0, compress=0)
writer.write(fout)
hres = open(fres, mode='w', encoding='utf-8',
errors='replace')
hres.write(self.msg)
hres.close()
# let's see if we have any allowed exception file
fdif = os.path.join(TEST_DIR, (file_name + ".difs"))
try:
hdif = open(fdif)
msg = hdif.read()
hdif.close()
except (FileNotFoundError, IOError):
msg = ""
# if exception file matches exactly, we are done.
if self.msg != msg:
self.msg = "\n****Captured Output****\n" + output[0] + \
"\n****Captured Err****\n" + output[1] + \
"\n****End Capture Err****\n" + self.msg
self.fail(self.msg)
return tst
# let's see if we have a single file to run, example;