mirror of
https://github.com/gryf/ebook-converter.git
synced 2026-01-11 14:24:12 +01:00
Removed unused startuo module
This commit is contained in:
@@ -1,154 +0,0 @@
|
||||
"""
|
||||
Perform various initialization tasks.
|
||||
"""
|
||||
import builtins
|
||||
import fcntl
|
||||
import locale
|
||||
import sys
|
||||
import threading
|
||||
|
||||
from ebook_converter import constants_old
|
||||
|
||||
# For backwards compat with some third party plugins
|
||||
builtins.__dict__['dynamic_property'] = lambda func: func(None)
|
||||
|
||||
|
||||
_run_once = False
|
||||
|
||||
if not _run_once:
|
||||
_run_once = True
|
||||
|
||||
# Ensure that all temp files/dirs are created under a calibre tmp dir
|
||||
from ebook_converter.ptempfile import base_dir
|
||||
try:
|
||||
base_dir()
|
||||
except EnvironmentError:
|
||||
# Ignore this error during startup, so we can show a better error
|
||||
# message to the user later.
|
||||
pass
|
||||
|
||||
#
|
||||
# Convert command line arguments to unicode
|
||||
enc = constants_old.preferred_encoding
|
||||
for i in range(1, len(sys.argv)):
|
||||
if not isinstance(sys.argv[i], str):
|
||||
sys.argv[i] = sys.argv[i].decode(enc, 'replace')
|
||||
|
||||
#
|
||||
# Ensure that the max number of open files is at least 1024
|
||||
|
||||
import resource
|
||||
soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
|
||||
if soft < 1024:
|
||||
try:
|
||||
resource.setrlimit(resource.RLIMIT_NOFILE, (min(1024, hard), hard))
|
||||
except Exception:
|
||||
if constants_old.DEBUG:
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
#
|
||||
# Setup resources
|
||||
import ebook_converter.utils.resources as resources
|
||||
resources
|
||||
|
||||
# Initialize locale
|
||||
# Import string as we do not want locale specific
|
||||
# string.whitespace/printable, on windows especially, this causes problems.
|
||||
# Before the delay load optimizations, string was loaded before this point
|
||||
# anyway, so we preserve the old behavior explicitly.
|
||||
import string
|
||||
string
|
||||
try:
|
||||
# set the locale to the user's default locale
|
||||
locale.setlocale(locale.LC_ALL, '')
|
||||
except Exception:
|
||||
dl = locale.getdefaultlocale()
|
||||
try:
|
||||
if dl:
|
||||
locale.setlocale(locale.LC_ALL, dl[0])
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def connect_lambda(bound_signal, self, func, **kw):
|
||||
import weakref
|
||||
r = weakref.ref(self)
|
||||
del self
|
||||
num_args = func.__code__.co_argcount - 1
|
||||
if num_args < 0:
|
||||
raise TypeError('lambda must take at least one argument')
|
||||
|
||||
def slot(*args):
|
||||
ctx = r()
|
||||
if ctx is not None:
|
||||
if len(args) != num_args:
|
||||
args = args[:num_args]
|
||||
func(ctx, *args)
|
||||
|
||||
bound_signal.connect(slot, **kw)
|
||||
builtins.__dict__['connect_lambda'] = connect_lambda
|
||||
|
||||
# Name all threads at the OS level created using the threading module, see
|
||||
# http://bugs.python.org/issue15500
|
||||
|
||||
orig_start = threading.Thread.start
|
||||
|
||||
def new_start(self):
|
||||
orig_start(self)
|
||||
try:
|
||||
name = self.name
|
||||
if not name or name.startswith('Thread-'):
|
||||
name = self.__class__.__name__
|
||||
if name == 'Thread':
|
||||
name = self.name
|
||||
if name:
|
||||
if isinstance(name, str):
|
||||
name = name.encode('ascii', 'replace').decode('ascii')
|
||||
constants_old.plugins['speedup'][0].set_thread_name(name[:15])
|
||||
except Exception:
|
||||
pass # Don't care about failure to set name
|
||||
threading.Thread.start = new_start
|
||||
|
||||
|
||||
def test_lopen():
|
||||
from ebook_converter.ptempfile import TemporaryDirectory
|
||||
from ebook_converter import CurrentDir
|
||||
n = 'f\xe4llen'
|
||||
print('testing open()')
|
||||
|
||||
def assert_not_inheritable(f):
|
||||
if not fcntl.fcntl(f, fcntl.F_GETFD) & fcntl.FD_CLOEXEC:
|
||||
raise SystemExit('File handle is inheritable!')
|
||||
|
||||
def copen(*args):
|
||||
ans = open(*args)
|
||||
assert_not_inheritable(ans)
|
||||
return ans
|
||||
|
||||
with TemporaryDirectory() as tdir, CurrentDir(tdir):
|
||||
with copen(n, 'w') as f:
|
||||
f.write('one')
|
||||
|
||||
print('O_CREAT tested')
|
||||
with copen(n, 'w+b') as f:
|
||||
f.write(b'two')
|
||||
with copen(n, 'r') as f:
|
||||
if f.read() == 'two':
|
||||
print('O_TRUNC tested')
|
||||
else:
|
||||
raise Exception('O_TRUNC failed')
|
||||
with copen(n, 'ab') as f:
|
||||
f.write(b'three')
|
||||
with copen(n, 'r+') as f:
|
||||
if f.read() == 'twothree':
|
||||
print('O_APPEND tested')
|
||||
else:
|
||||
raise Exception('O_APPEND failed')
|
||||
with copen(n, 'r+') as f:
|
||||
f.seek(3)
|
||||
f.write('xxxxx')
|
||||
f.seek(0)
|
||||
if f.read() == 'twoxxxxx':
|
||||
print('O_RANDOM tested')
|
||||
else:
|
||||
raise Exception('O_RANDOM failed')
|
||||
Reference in New Issue
Block a user