bn8/python-mylib
1
0
Fork 0
Code Issues Pull requests Projects Releases 12 Wiki Activity Actions

Code cleaning

Browse source
This commit is contained in:
Benjamin Renard 2021-05-19 19:19:57 +02:00
parent ad31ddc699
commit 6bbacce38a
15 changed files with 186 additions and 124 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
.pylintrc Normal file
View file

@ -0,0 +1,6 @@
[MESSAGES CONTROL]
disable=invalid-name,
locally-disabled,
too-many-arguments,
too-many-branches,
line-too-long,

2
mylib/email.py
View file

@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
""" Email helpers """
""" Email client to forge and send emails """
import logging
import os

62
mylib/ldap.py
View file

@ -1,4 +1,6 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
""" LDAP server connection helper """
import copy
import datetime
@ -12,7 +14,9 @@ from ldap.controls.simple import RelaxRulesControl
import ldap.modlist as modlist
import pytz
class LdapServer(object): # pylint: disable=useless-object-inheritance
class LdapServer:
""" LDAP server connection helper """ # pylint: disable=useless-object-inheritance
uri = None
dn = None
@ -21,24 +25,25 @@ class LdapServer(object): # pylint: disable=useless-object-inheritance
con = 0
def __init__(self,uri,dn=None,pwd=None,v2=None,raiseOnError=False, logger=False):
self.uri = uri
self.dn = dn
self.pwd = pwd
self.raiseOnError = raiseOnError
def __init__(self, uri, dn=None, pwd=None, v2=None, raiseOnError=False, logger=False):
self.uri = uri
self.dn = dn
self.pwd = pwd
self.raiseOnError = raiseOnError
if v2:
self.v2=True
self.v2 = True
if logger:
self.logger = logger
else:
self.logger = logging.getLogger(__name__)
def _error(self,error,level=logging.WARNING):
def _error(self, error, level=logging.WARNING):
if self.raiseOnError:
raise LdapServerException(error)
self.logger.log(level, error)
def connect(self):
""" Start connection to LDAP server """
if self.con == 0:
try:
con = ldap.initialize(self.uri)
@ -48,19 +53,20 @@ class LdapServer(object): # pylint: disable=useless-object-inheritance
con.protocol_version = ldap.VERSION3 # pylint: disable=no-member
if self.dn:
con.simple_bind_s(self.dn,self.pwd)
con.simple_bind_s(self.dn, self.pwd)
elif self.uri.startswith('ldapi://'):
con.sasl_interactive_bind_s("", ldap.sasl.external())
self.con = con
return True
except ldap.LDAPError as e: # pylint: disable=no-member
self._error('LdapServer - Error connecting and binding to LDAP server : %s' % e,logging.CRITICAL)
self._error('LdapServer - Error connecting and binding to LDAP server : %s' % e, logging.CRITICAL)
return False
return True
@staticmethod
def get_scope(scope):
""" Map scope parameter to python-ldap value """
if scope == 'base':
return ldap.SCOPE_BASE # pylint: disable=no-member
if scope == 'one':
@ -70,6 +76,7 @@ class LdapServer(object): # pylint: disable=useless-object-inheritance
raise Exception("Unknown LDAP scope '%s'" % scope)
def search(self, basedn, filterstr=None, attrs=None, sizelimit=0, scope=None):
""" Run a search on LDAP server """
res_id = self.con.search(
basedn,
self.get_scope(scope if scope else 'sub'),
@ -79,7 +86,7 @@ class LdapServer(object): # pylint: disable=useless-object-inheritance
ret = {}
c = 0
while True:
res_type, res_data = self.con.result(res_id,0)
res_type, res_data = self.con.result(res_id, 0)
if res_data == [] or (sizelimit and c > sizelimit):
break
if res_type == ldap.RES_SEARCH_ENTRY: # pylint: disable=no-member
@ -88,10 +95,12 @@ class LdapServer(object): # pylint: disable=useless-object-inheritance
return ret
def get_object(self, dn, filterstr=None, attrs=None):
""" Retrieve a LDAP object specified by its DN """
result = self.search(dn, filterstr=filterstr, scope='base', attrs=attrs)
return result[dn] if dn in result else None
def paged_search(self, basedn, filterstr, attrs, scope='sub', pagesize=500):
""" Run a paged search on LDAP server """
assert not self.v2, "Paged search is not available on LDAP version 2"
# Initialize SimplePagedResultsControl object
page_control = SimplePagedResultsControl(
@ -161,18 +170,20 @@ class LdapServer(object): # pylint: disable=useless-object-inheritance
self.logger.debug("LdapServer - Paged search end: %d object(s) retreived in %d page(s) of %d object(s)", len(ret), pages_count, pagesize)
return ret
def add_object(self,dn,attrs):
def add_object(self, dn, attrs):
""" Add an object in LDAP directory """
ldif = modlist.addModlist(attrs)
try:
self.logger.debug("LdapServer - Add %s", dn)
self.con.add_s(dn,ldif)
self.con.add_s(dn, ldif)
return True
except ldap.LDAPError as e: # pylint: disable=no-member
self._error("LdapServer - Error adding %s : %s" % (dn,e), logging.ERROR)
self._error("LdapServer - Error adding %s : %s" % (dn, e), logging.ERROR)
return False
def update_object(self, dn, old, new, ignore_attrs=None, relax=False):
""" Update an object in LDAP directory """
assert not relax or not self.v2, "Relax modification is not available on LDAP version 2"
ldif = modlist.modifyModlist(
old, new,
@ -192,6 +203,7 @@ class LdapServer(object): # pylint: disable=useless-object-inheritance
@staticmethod
def update_need(old, new, ignore_attrs=None):
""" Check if an update is need on a LDAP object based on its old and new attributes values """
ldif = modlist.modifyModlist(
old, new,
ignore_attr_types=ignore_attrs if ignore_attrs else []
@ -202,6 +214,7 @@ class LdapServer(object): # pylint: disable=useless-object-inheritance
@staticmethod
def get_changes(old, new, ignore_attrs=None):
""" Retrieve changes (as modlist) on an object based on its old and new attributes values """
return modlist.modifyModlist(
old, new,
ignore_attr_types=ignore_attrs if ignore_attrs else []
@ -209,6 +222,7 @@ class LdapServer(object): # pylint: disable=useless-object-inheritance
@staticmethod
def format_changes(old, new, ignore_attrs=None, prefix=None):
""" Format changes (modlist) on an object based on its old and new attributes values to display/log it """
msg = []
for (op, attr, val) in modlist.modifyModlist(old, new, ignore_attr_types=ignore_attrs if ignore_attrs else []):
if op == ldap.MOD_ADD: # pylint: disable=no-member
@ -226,6 +240,7 @@ class LdapServer(object): # pylint: disable=useless-object-inheritance
return '\n'.join(msg)
def rename_object(self, dn, new_rdn, new_sup=None, delete_old=True):
""" Rename an object in LDAP directory """
# If new_rdn is a complete DN, split new RDN and new superior DN
if len(new_rdn.split(',')) > 1:
self.logger.debug(
@ -261,21 +276,24 @@ class LdapServer(object): # pylint: disable=useless-object-inheritance
return False
def drop_object(self, dn):
""" Drop an object in LDAP directory """
try:
self.logger.debug("LdapServer - Delete %s", dn)
self.con.delete_s(dn)
return True
except ldap.LDAPError as e: # pylint: disable=no-member
self._error("LdapServer - Error deleting %s : %s" % (dn,e), logging.ERROR)
self._error("LdapServer - Error deleting %s : %s" % (dn, e), logging.ERROR)
return False
@staticmethod
def get_dn(obj):
""" Retreive an on object DN from its entry in LDAP search result """
return obj[0][0]
@staticmethod
def get_attr(obj, attr, all=None, default=None):
""" Retreive an on object attribute value(s) from the object entry in LDAP search result """
if attr not in obj:
for k in obj:
if k.lower() == attr.lower():
@ -289,13 +307,18 @@ class LdapServer(object): # pylint: disable=useless-object-inheritance
return obj[attr][0]
return default
class LdapServerException(BaseException):
def __init__(self,msg):
""" Generic exception raised by LdapServer """
def __init__(self, msg):
BaseException.__init__(self, msg)
#
# Helpers
# LDAP date string helpers
#
def parse_datetime(value, to_timezone=None, default_timezone=None, naive=None):
"""
Convert LDAP date string to datetime.datetime object
@ -335,6 +358,7 @@ def parse_datetime(value, to_timezone=None, default_timezone=None, naive=None):
return date.astimezone(to_timezone)
return date
def parse_date(value, to_timezone=None, default_timezone=None, naive=None):
"""
Convert LDAP date string to datetime.date object
@ -348,6 +372,7 @@ def parse_date(value, to_timezone=None, default_timezone=None, naive=None):
"""
return parse_datetime(value, to_timezone, default_timezone, naive).date()
def format_datetime(value, from_timezone=None, to_timezone=None, naive=None):
"""
Convert datetime.datetime object to LDAP date string
@ -388,6 +413,7 @@ def format_datetime(value, from_timezone=None, to_timezone=None, naive=None):
datestring = datestring.replace('+0000', 'Z')
return datestring
def format_date(value, from_timezone=None, to_timezone=None, naive=None):
"""
Convert datetime.date object to LDAP date string

11
mylib/mysql.py
View file

@ -9,21 +9,22 @@ import MySQLdb
log = logging.getLogger(__name__)
class MyDB:
""" MySQL client """
host = ""
user = ""
pwd = ""
db = ""
pwd = ""
db = ""
con = 0
def __init__(self, host, user, pwd, db):
self.host = host
self.user = user
self.pwd = pwd
self.db = db
self.pwd = pwd
self.db = db
def connect(self):
""" Connect to MySQL server """
@ -35,7 +36,7 @@ class MyDB:
log.fatal('Error connecting to MySQL server', exc_info=True)
sys.exit(1)
def doSQL(self,sql):
def doSQL(self, sql):
""" Run INSERT/UPDATE/DELETE/... SQL query """
cursor = self.con.cursor()
try:

148
mylib/opening_hours.py
View file

@ -1,3 +1,7 @@
# -*- coding: utf-8 -*-
""" Opening hours helpers """
import datetime
import re
import time
@ -10,37 +14,39 @@ date_format = '%d/%m/%Y'
date_pattern = re.compile('^([0-9]{2})/([0-9]{2})/([0-9]{4})$')
time_pattern = re.compile('^([0-9]{1,2})h([0-9]{2})?$')
def easter_date(year):
a = year//100
b = year%100
c = (3*(a+25))//4
d = (3*(a+25))%4
e = (8*(a+11))//25
f = (5*a+b)%19
g = (19*f+c-e)%30
h = (f+11*g)//319
j = (60*(5-d)+b)//4
k = (60*(5-d)+b)%4
m = (2*j-k-g+h)%7
n = (g-h+m+114)//31
p = (g-h+m+114)%31
day = p+1
a = year // 100
b = year % 100
c = (3 * (a + 25)) // 4
d = (3 * (a + 25)) % 4
e = (8 * (a + 11)) // 25
f = (5 * a + b) % 19
g = (19 * f + c - e) % 30
h = (f + 11 * g) // 319
j = (60 * (5 - d) + b) // 4
k = (60 * (5 - d) + b) % 4
m = (2 * j - k - g + h) % 7
n = (g - h + m + 114) // 31
p = (g - h + m + 114) % 31
day = p + 1
month = n
return datetime.date(year, month, day)
def nonworking_french_public_days_of_the_year(year=None):
if year is None:
year=datetime.date.today().year
dp=easter_date(year)
year = datetime.date.today().year
dp = easter_date(year)
return {
'1janvier': datetime.date(year, 1, 1),
'paques': dp,
'lundi_paques': (dp+datetime.timedelta(1)),
'lundi_paques': (dp + datetime.timedelta(1)),
'1mai': datetime.date(year, 5, 1),
'8mai': datetime.date(year, 5, 8),
'jeudi_ascension': (dp+datetime.timedelta(39)),
'pentecote': (dp+datetime.timedelta(49)),
'lundi_pentecote': (dp+datetime.timedelta(50)),
'jeudi_ascension': (dp + datetime.timedelta(39)),
'pentecote': (dp + datetime.timedelta(49)),
'lundi_pentecote': (dp + datetime.timedelta(50)),
'14juillet': datetime.date(year, 7, 14),
'15aout': datetime.date(year, 8, 15),
'1novembre': datetime.date(year, 11, 1),
@ -49,8 +55,9 @@ def nonworking_french_public_days_of_the_year(year=None):
'saint_etienne': datetime.date(year, 12, 26),
}
def parse_exceptional_closures(values):
exceptional_closures=[]
exceptional_closures = []
for value in values:
days = []
hours_periods = []
@ -60,15 +67,15 @@ def parse_exceptional_closures(values):
continue
parts = word.split('-')
if len(parts) == 1:
# ex : 31/02/2017
# ex: 31/02/2017
ptime = time.strptime(word, date_format)
date = datetime.date(ptime.tm_year, ptime.tm_mon, ptime.tm_mday)
if date not in days:
days.append(date)
elif len(parts) == 2:
# ex : 18/12/2017-20/12/2017 ou 9h-10h30
# ex: 18/12/2017-20/12/2017 ou 9h-10h30
if date_pattern.match(parts[0]) and date_pattern.match(parts[1]):
# ex : 18/12/2017-20/12/2017
# ex: 18/12/2017-20/12/2017
pstart = time.strptime(parts[0], date_format)
pstop = time.strptime(parts[1], date_format)
if pstop <= pstart:
@ -81,7 +88,7 @@ def parse_exceptional_closures(values):
days.append(date)
date += datetime.timedelta(days=1)
else:
# ex : 9h-10h30
# ex: 9h-10h30
mstart = time_pattern.match(parts[0])
mstop = time_pattern.match(parts[1])
if not mstart or not mstop:
@ -92,7 +99,7 @@ def parse_exceptional_closures(values):
raise ValueError('Time %s <= %s' % (parts[1], parts[0]))
hours_periods.append({'start': hstart, 'stop': hstop})
else:
raise ValueError('Invalid number of part in this word : "%s"' % word)
raise ValueError('Invalid number of part in this word: "%s"' % word)
if not days:
raise ValueError('No days found in value "%s"' % value)
exceptional_closures.append({'days': days, 'hours_periods': hours_periods})
@ -100,82 +107,85 @@ def parse_exceptional_closures(values):
def parse_normal_opening_hours(values):
normal_opening_hours=[]
normal_opening_hours = []
for value in values:
days=[]
hours_periods=[]
words=value.strip().split()
days = []
hours_periods = []
words = value.strip().split()
for word in words:
if word=='':
if not word:
continue
parts=word.split('-')
if len(parts)==1:
# ex : jeudi
parts = word.split('-')
if len(parts) == 1:
# ex: jeudi
if word not in week_days:
raise ValueError('"%s" is not a valid week day' % word)
if word not in days:
days.append(word)
elif len(parts)==2:
# ex : lundi-jeudi ou 9h-10h30
elif len(parts) == 2:
# ex: lundi-jeudi ou 9h-10h30
if parts[0] in week_days and parts[1] in week_days:
# ex : lundi-jeudi
# ex: lundi-jeudi
if week_days.index(parts[1]) <= week_days.index(parts[0]):
raise ValueError('"%s" is before "%s"' % (parts[1],parts[0]))
started=False
raise ValueError('"%s" is before "%s"' % (parts[1], parts[0]))
started = False
for d in week_days:
if not started and d!=parts[0]:
if not started and d != parts[0]:
continue
started=True
started = True
if d not in days:
days.append(d)
if d==parts[1]:
if d == parts[1]:
break
else:
#ex : 9h-10h30
mstart=time_pattern.match(parts[0])
mstop=time_pattern.match(parts[1])
# ex: 9h-10h30
mstart = time_pattern.match(parts[0])
mstop = time_pattern.match(parts[1])
if not mstart or not mstop:
raise ValueError('"%s" is not a valid time period' % word)
hstart=datetime.time(int(mstart.group(1)), int(mstart.group(2) or 0))
hstop=datetime.time(int(mstop.group(1)), int(mstop.group(2) or 0))
if hstop<=hstart:
raise ValueError('Time %s <= %s' % (parts[1],parts[0]))
hstart = datetime.time(int(mstart.group(1)), int(mstart.group(2) or 0))
hstop = datetime.time(int(mstop.group(1)), int(mstop.group(2) or 0))
if hstop <= hstart:
raise ValueError('Time %s <= %s' % (parts[1], parts[0]))
hours_periods.append({'start': hstart, 'stop': hstop})
else:
raise ValueError('Invalid number of part in this word : "%s"' % word)
raise ValueError('Invalid number of part in this word: "%s"' % word)
if not days and not hours_periods:
raise ValueError('No days or hours period found in this value : "%s"' % value)
raise ValueError('No days or hours period found in this value: "%s"' % value)
normal_opening_hours.append({'days': days, 'hours_periods': hours_periods})
return normal_opening_hours
def is_closed(normal_opening_hours_values=None, exceptional_closures_values=None,
nonworking_public_holidays_values=None, exceptional_closure_on_nonworking_public_days=False,
when=None, on_error='raise'):
def is_closed(
normal_opening_hours_values=None, exceptional_closures_values=None,
nonworking_public_holidays_values=None, exceptional_closure_on_nonworking_public_days=False,
when=None, on_error='raise'
):
if not when:
when = datetime.datetime.now()
when_date=when.date()
when_time=when.time()
when_weekday=week_days[when.timetuple().tm_wday]
on_error_result=None
if on_error=='closed':
on_error_result={'closed': True, 'exceptional_closure': False, 'exceptional_closure_all_day': False}
elif on_error=='opened':
on_error_result={'closed': False, 'exceptional_closure': False, 'exceptional_closure_all_day': False}
when_date = when.date()
when_time = when.time()
when_weekday = week_days[when.timetuple().tm_wday]
on_error_result = None
if on_error == 'closed':
on_error_result = {'closed': True, 'exceptional_closure': False, 'exceptional_closure_all_day': False}
elif on_error == 'opened':
on_error_result = {'closed': False, 'exceptional_closure': False, 'exceptional_closure_all_day': False}
log.debug("When = %s => date = %s / time = %s / week day = %s", when, when_date, when_time, when_weekday)
if nonworking_public_holidays_values:
log.debug("Nonworking public holidays: %s", nonworking_public_holidays_values)
nonworking_days=nonworking_french_public_days_of_the_year()
nonworking_days = nonworking_french_public_days_of_the_year()
for day in nonworking_public_holidays_values:
if day in nonworking_days and when_date==nonworking_days[day]:
if day in nonworking_days and when_date == nonworking_days[day]:
log.debug("Non working day: %s", day)
return {'closed': True, 'exceptional_closure': exceptional_closure_on_nonworking_public_days, 'exceptional_closure_all_day': exceptional_closure_on_nonworking_public_days}
if len(exceptional_closures_values)>0:
if exceptional_closures_values:
try:
exceptional_closures=parse_exceptional_closures(exceptional_closures_values)
exceptional_closures = parse_exceptional_closures(exceptional_closures_values)
log.debug('Exceptional closures: %s', exceptional_closures)
except Exception as e:
except ValueError as e:
log.error("Fail to parse exceptional closures, consider as closed", exc_info=True)
if on_error_result is None:
raise e from e
@ -193,9 +203,9 @@ def is_closed(normal_opening_hours_values=None, exceptional_closures_values=None
if normal_opening_hours_values:
try:
normal_opening_hours=parse_normal_opening_hours(normal_opening_hours_values)
normal_opening_hours = parse_normal_opening_hours(normal_opening_hours_values)
log.debug('Normal opening hours: %s', normal_opening_hours)
except Exception:
except ValueError as e: # pylint: disable=broad-except
log.error("Fail to parse normal opening hours, consider as closed", exc_info=True)
if on_error_result is None:
raise e from e

3
mylib/pbar.py
View file

@ -8,7 +8,8 @@ import progressbar
log = logging.getLogger(__name__)
class Pbar(object): # pylint: disable=useless-object-inheritance
class Pbar: # pylint: disable=useless-object-inheritance
"""
Progress bar

23
mylib/pgsql.py
View file

@ -10,13 +10,14 @@ import psycopg2
log = logging.getLogger(__name__)
class PgDB:
""" PostgreSQL client """
host = ""
user = ""
pwd = ""
db = ""
pwd = ""
db = ""
con = 0
@ -34,7 +35,11 @@ class PgDB:
""" Connect to PostgreSQL server """
if self.con == 0:
try:
con = psycopg2.connect("dbname='%s' user='%s' host='%s' password='%s'" % (self.db,self.user,self.host,self.pwd))
con = psycopg2.connect(
"dbname='%s' user='%s' host='%s' password='%s'" % (
self.db, self.user, self.host, self.pwd
)
)
self.con = con
except Exception:
log.fatal('An error occured during Postgresql database connection.', exc_info=1)
@ -131,7 +136,7 @@ class PgDB:
def insert(self, table, values, just_try=False):
""" Run INSERT SQL query """
sql=u"INSERT INTO %s (%s) VALUES (%s)" % (table, u', '.join(values.keys()), u", ".join(map(lambda x: self._quote_value(values[x]), values)))
sql = u"INSERT INTO %s (%s) VALUES (%s)" % (table, u', '.join(values.keys()), u", ".join(map(lambda x: self._quote_value(values[x]), values)))
if just_try:
log.debug(u"Just-try mode : execute INSERT query : %s", sql)
@ -145,11 +150,11 @@ class PgDB:
def update(self, table, values, where_clauses, where_op=u'AND', just_try=False):
""" Run UPDATE SQL query """
where=self._format_where_clauses(where_clauses, where_op=where_op)
where = self._format_where_clauses(where_clauses, where_op=where_op)
if not where:
return False
sql=u"UPDATE %s SET %s WHERE %s" % (table, u", ".join(map(lambda x: "%s=%s" % (x, self._quote_value(values[x])), values)), where)
sql = u"UPDATE %s SET %s WHERE %s" % (table, u", ".join(map(lambda x: "%s=%s" % (x, self._quote_value(values[x])), values)), where)
if just_try:
log.debug(u"Just-try mode : execute UPDATE query : %s", sql)
@ -163,11 +168,11 @@ class PgDB:
def delete(self, table, where_clauses, where_op=u'AND', just_try=False):
""" Run DELETE SQL query """
where=self._format_where_clauses(where_clauses, where_op=where_op)
where = self._format_where_clauses(where_clauses, where_op=where_op)
if not where:
return False
sql=u"DELETE FROM %s WHERE %s" % (table, where)
sql = u"DELETE FROM %s WHERE %s" % (table, where)
if just_try:
log.debug(u"Just-try mode : execute DELETE query : %s", sql)
@ -191,7 +196,7 @@ class PgDB:
sql += u" FROM " + table
if where_clauses:
where=self._format_where_clauses(where_clauses, where_op=where_op)
where = self._format_where_clauses(where_clauses, where_op=where_op)
if not where:
return False

3
mylib/report.py
View file

@ -8,7 +8,8 @@ import logging
log = logging.getLogger(__name__)
class Report(object): # pylint: disable=useless-object-inheritance
class Report: # pylint: disable=useless-object-inheritance
""" Logging report """
content = []

5
mylib/scripts/email_test.py
View file

@ -5,8 +5,8 @@ import datetime
import logging
import sys
import argparse
import getpass
from mako.template import Template as MakoTemplate
from mylib.scripts.helpers import get_opts_parser, add_email_opts
from mylib.scripts.helpers import init_logging, init_email_client
@ -14,7 +14,8 @@ from mylib.scripts.helpers import init_logging, init_email_client
log = logging.getLogger('mylib.scripts.email_test')
def main(argv=None): #pylint: disable=too-many-locals,too-many-statements
def main(argv=None): # pylint: disable=too-many-locals,too-many-statements
""" Script main """
if argv is None:
argv = sys.argv[1:]

4
mylib/scripts/helpers.py
View file

@ -9,8 +9,9 @@ from mylib.email import EmailClient
log = logging.getLogger(__name__)
def init_logging(options, name, report=None):
""" Initialize logs """
""" Initialize logging from calling script options """
logformat = '%(asctime)s - ' + name + ' - %(levelname)s - %(message)s'
if options.debug:
loglevel = logging.DEBUG
@ -172,6 +173,7 @@ def add_email_opts(parser):
def init_email_client(options, **kwargs):
""" Initialize email client from calling script options """
log.info('Initialize Email client')
return EmailClient(
smtp_host=options.email_smtp_host,

9
mylib/scripts/ldap_test.py
View file

@ -8,14 +8,15 @@ import sys
import dateutil.tz
import pytz
from mylib.ldap import format_datetime,format_date, parse_datetime, parse_date
from mylib.ldap import format_datetime, format_date, parse_datetime, parse_date
from mylib.scripts.helpers import get_opts_parser
from mylib.scripts.helpers import init_logging
log = logging.getLogger('mylib.scripts.ldap_test')
def main(argv=None): #pylint: disable=too-many-locals,too-many-statements
def main(argv=None): # pylint: disable=too-many-locals,too-many-statements
""" Script main """
if argv is None:
argv = sys.argv[1:]
@ -24,6 +25,9 @@ def main(argv=None): #pylint: disable=too-many-locals,too-many-statements
parser = get_opts_parser(just_try=True)
options = parser.parse_args()
# Initialize logs
init_logging(options, 'Test LDAP helpers')
now = datetime.datetime.now().replace(tzinfo=dateutil.tz.tzlocal())
print("Now = %s" % now)
@ -50,7 +54,6 @@ def main(argv=None): #pylint: disable=too-many-locals,too-many-statements
print("format_date (to_timezone=Tokyo) : %s" % format_date(now, to_timezone='Asia/Tokyo'))
print("format_date (naive=True) : %s" % format_date(now, naive=True))
print("parse_datetime : %s" % parse_datetime(datestring_now))
print("parse_datetime (default_timezone=utc) : %s" % parse_datetime(datestring_now[0:-1], default_timezone=pytz.utc))
print("parse_datetime (default_timezone=local) : %s" % parse_datetime(datestring_now[0:-1], default_timezone=dateutil.tz.tzlocal()))

3
mylib/scripts/pbar_test.py
View file

@ -12,7 +12,8 @@ from mylib.scripts.helpers import init_logging
log = logging.getLogger('mylib.scripts.pbar_test')
def main(argv=None): #pylint: disable=too-many-locals,too-many-statements
def main(argv=None): # pylint: disable=too-many-locals,too-many-statements
""" Script main """
if argv is None:
argv = sys.argv[1:]

3
mylib/scripts/report_test.py
View file

@ -11,7 +11,8 @@ from mylib.scripts.helpers import init_logging, init_email_client
log = logging.getLogger('mylib.scripts.report_test')
def main(argv=None): #pylint: disable=too-many-locals,too-many-statements
def main(argv=None): # pylint: disable=too-many-locals,too-many-statements
""" Script main """
if argv is None:
argv = sys.argv[1:]

2
setup.cfg Normal file
View file

@ -0,0 +1,2 @@
[flake8]
ignore = E501

26
tests/test_opening_hours.py
View file

@ -43,6 +43,7 @@ def test_parse_exceptional_closures_full_days_period():
}
]
def test_parse_exceptional_closures_invalid_days_period():
with pytest.raises(ValueError):
opening_hours.parse_exceptional_closures(["22/09/2017-21/09/2017"])
@ -156,21 +157,22 @@ def test_parse_normal_opening_hours_multiple_periods():
# Tests on is_closed
#
exceptional_closures = ["22/09/2017", "20/09/2017-22/09/2017", "20/09/2017-22/09/2017 18/09/2017", "25/11/2017", "26/11/2017 9h30-12h30"]
normal_opening_hours = ["lundi-mardi jeudi 9h30-12h30 14h-16h30", "mercredi vendredi 9h30-12h30 14h-17h"]
nonworking_public_holidays = [
'1janvier',
'paques',
'lundi_paques',
'1mai',
'8mai',
'jeudi_ascension',
'lundi_pentecote',
'14juillet',
'15aout',
'1novembre',
'11novembre',
'noel',
'1janvier',
'paques',
'lundi_paques',
'1mai',
'8mai',
'jeudi_ascension',
'lundi_pentecote',
'14juillet',
'15aout',
'1novembre',
'11novembre',
'noel',
]