'
example_text = \
'This post has a web links https://somesite.net\n\nAnd some other text'
linked_text = add_web_links(example_text)
expected_text = \
'https://' + \
'somesite.net'
if expected_text not in linked_text:
print(expected_text + '\n')
print(linked_text)
assert expected_text in linked_text
# NOTE: it is difficult to find academic studies of the fediverse which
# do not in some way violate consent or embody an arrogant status
# quo attitude. Did all those scraped accounts agree to be part of
# an academic study? Did they even consider consent as an issue?
# It seems doubtful. We are just like algae under a microscope to them.
example_text = \
'This post has an arxiv link arXiv:2203.15752 some other text'
linked_text = add_web_links(example_text)
expected_text = \
'arXiv:2203.15752'
if expected_text not in linked_text:
print(expected_text + '\n')
print(linked_text)
assert expected_text in linked_text
example_text = \
'This post has an doi link ' + \
'doi:10.1109/INFCOMW.2019.8845221 some other text'
linked_text = add_web_links(example_text)
expected_text = \
'' + \
'doi:10.1109/INFCOMW.2019.8845221'
if expected_text not in linked_text:
print(expected_text + '\n')
print(linked_text)
assert expected_text in linked_text
example_text = \
'This post has a very long web link\n\nhttp://' + \
'cbwebewuvfuftdiudbqd33dddbbyuef23fyug3bfhcyu2fct2' + \
'cuyqbcbucuwvckiwyfgewfvqejbchevbhwevuevwbqebqekve' + \
'qvuvjfkf.onion\n\nAnd some other text'
linked_text = add_web_links(example_text)
assert 'ellipsis' in linked_text
example_text = \
'
ABCABCABCABCABCABCABCABCABCABCABCABCABCA<\p>'
example_text = \
'"the nucleus of mutual-support institutions, habits, and customs ' + \
'remains alive with the millions; it keeps them together; and ' + \
'they prefer to cling to their customs, beliefs, and traditions ' + \
'rather than to accept the teachings of a war of each ' + \
'against all"\n\n--Peter Kropotkin'
test_fn_str = add_web_links(example_text)
result_text = remove_long_words(test_fn_str, 40, [])
assert result_text == example_text
assert 'ellipsis' not in result_text
example_text = \
'
'
assert not dangerous_markup(content, allow_local_network_access)
content = 'This is a valid message without markup'
assert not dangerous_markup(content, allow_local_network_access)
content = '
This is a valid-looking message. But wait... ' + \
'
'
if reply['object']['content'] != expected_str:
print(expected_str + '\n')
print(reply['object']['content'])
assert reply['object']['content'] == expected_str
reply['object']['contentMap'][system_language] = reply['object']['content']
assert reply['object']['tag'][0]['type'] == 'Mention'
assert reply['object']['tag'][0]['name'] == '@ninjarodent@rat.site'
assert reply['object']['tag'][0]['href'] == \
'https://rat.site/users/ninjarodent'
assert len(reply['object']['to']) == 1
assert reply['object']['to'][0].endswith('#Public')
assert len(reply['object']['cc']) >= 1
assert reply['object']['cc'][0].endswith(nickname + '/followers')
assert len(reply['object']['tag']) == 1
if len(reply['object']['cc']) != 2:
print('reply["object"]["cc"]: ' + str(reply['object']['cc']))
assert len(reply['object']['cc']) == 2
assert reply['object']['cc'][1] == \
http_prefix + '://rat.site/users/ninjarodent'
assert len(reply['to']) == 1
assert reply['to'][0].endswith('#Public')
assert len(reply['cc']) >= 1
assert reply['cc'][0].endswith(nickname + '/followers')
if len(reply['cc']) != 2:
print('reply["cc"]: ' + str(reply['cc']))
assert len(reply['cc']) == 2
assert reply['cc'][1] == http_prefix + '://rat.site/users/ninjarodent'
def _get_function_call_args(name: str, lines: [], start_line_ctr: int) -> []:
"""Returns the arguments of a function call given lines
of source code and a starting line number
"""
args_str = lines[start_line_ctr].split(name + '(')[1]
if ')' in args_str:
args_str = args_str.split(')')[0].replace(' ', '').split(',')
return args_str
for line_ctr in range(start_line_ctr + 1, len(lines)):
if ')' not in lines[line_ctr]:
args_str += lines[line_ctr]
continue
args_str += lines[line_ctr].split(')')[0]
break
return remove_eol(args_str).replace(' ', '').split(',')
def get_function_calls(name: str, lines: [], start_line_ctr: int,
function_properties: {}) -> []:
"""Returns the functions called by the given one,
Starting with the given source code at the given line
"""
calls_functions = []
function_content_str = ''
for line_ctr in range(start_line_ctr + 1, len(lines)):
line_str = lines[line_ctr].strip()
if line_str.startswith('def '):
break
if line_str.startswith('class '):
break
function_content_str += lines[line_ctr]
for func_name, _ in function_properties.items():
if func_name + '(' in function_content_str:
calls_functions.append(func_name)
return calls_functions
def _function_args_match(call_args: [], func_args: []) -> bool:
"""Do the function arguments match the function call arguments?
"""
if len(call_args) == len(func_args):
return True
# count non-optional arguments in function call
call_args_ctr = 0
for arg1 in call_args:
if arg1 == 'self':
continue
if '=' not in arg1 or arg1.startswith("'"):
call_args_ctr += 1
# count non-optional arguments in function def
func_args_ctr = 0
for arg2 in func_args:
if arg2 == 'self':
continue
if '=' not in arg2 or arg2.startswith("'"):
func_args_ctr += 1
return call_args_ctr >= func_args_ctr
def _module_in_groups(mod_name: str, include_groups: [],
mod_groups: {}) -> bool:
"""Is the given module within the included groups list?
"""
for group_name in include_groups:
if mod_name in mod_groups[group_name]:
return True
return False
def _diagram_groups(include_groups: [],
exclude_extra_modules: [],
modules: {}, mod_groups: {},
max_module_calls: int) -> None:
"""Draws a dot diagram containing only the given module groups
"""
call_graph_str = 'digraph EpicyonGroups {\n\n'
call_graph_str += \
' graph [fontsize=10 fontname="Verdana" compound=true];\n'
call_graph_str += ' node [fontsize=10 fontname="Verdana"];\n\n'
exclude_modules_from_diagram = [
'setup', 'tests', '__init__', 'pyjsonld'
]
exclude_modules_from_diagram += exclude_extra_modules
# colors of modules nodes
for mod_name, mod_properties in modules.items():
if mod_name in exclude_modules_from_diagram:
continue
if not _module_in_groups(mod_name, include_groups, mod_groups):
continue
if not mod_properties.get('calls'):
call_graph_str += ' "' + mod_name + \
'" [fillcolor=yellow style=filled];\n'
continue
if len(mod_properties['calls']) <= int(max_module_calls / 8):
call_graph_str += ' "' + mod_name + \
'" [fillcolor=green style=filled];\n'
elif len(mod_properties['calls']) < int(max_module_calls / 4):
call_graph_str += ' "' + mod_name + \
'" [fillcolor=orange style=filled];\n'
else:
call_graph_str += ' "' + mod_name + \
'" [fillcolor=red style=filled];\n'
call_graph_str += '\n'
# connections between modules
for mod_name, mod_properties in modules.items():
if mod_name in exclude_modules_from_diagram:
continue
if not _module_in_groups(mod_name, include_groups, mod_groups):
continue
if not mod_properties.get('calls'):
continue
for mod_call in mod_properties['calls']:
if mod_call in exclude_modules_from_diagram:
continue
if not _module_in_groups(mod_call, include_groups, mod_groups):
continue
call_graph_str += ' "' + mod_name + '" -> "' + mod_call + '";\n'
# module groups/clusters
cluster_ctr = 1
for group_name, group_modules in mod_groups.items():
if group_name not in include_groups:
continue
call_graph_str += '\n'
call_graph_str += \
' subgraph cluster_' + str(cluster_ctr) + ' {\n'
call_graph_str += ' node [style=filled];\n'
for mod_name in group_modules:
if mod_name not in exclude_modules_from_diagram:
call_graph_str += ' ' + mod_name + ';\n'
call_graph_str += ' label = "' + group_name + '";\n'
call_graph_str += ' color = blue;\n'
call_graph_str += ' }\n'
cluster_ctr += 1
call_graph_str += '\n}\n'
filename = 'epicyon_groups'
for group_name in include_groups:
filename += '_' + group_name.replace(' ', '-')
filename += '.dot'
with open(filename, 'w+', encoding='utf-8') as fp_graph:
fp_graph.write(call_graph_str)
print('Graph saved to ' + filename)
print('Plot using: ' +
'sfdp -x -Goverlap=false -Goverlap_scaling=2 ' +
'-Gsep=+100 -Tx11 epicyon_modules.dot')
def _test_post_variable_names():
print('test_post_variable_names')
for _, _, files in os.walk('.'):
for source_file in files:
if not source_file.endswith('.py'):
continue
if source_file.startswith('.#'):
continue
if source_file.startswith('flycheck_'):
continue
source_str = ''
with open(source_file, 'r', encoding='utf-8') as file_source:
source_str = file_source.read()
if not source_str:
continue
if ' name="' not in source_str:
continue
names_list = source_str.split(' name="')
for index in range(1, len(names_list)):
if '"' not in names_list[index]:
continue
name_var = names_list[index].split('"')[0]
if not name_var:
continue
if ' ' in name_var:
continue
if '_' in name_var:
print(name_var + ' is not camel case POST variable in ' +
source_file)
assert False
break
def _test_config_param_names():
print('test_config_param_names')
fnames = ('get_config_param', 'set_config_param')
for _, _, files in os.walk('.'):
for source_file in files:
if not source_file.endswith('.py'):
continue
if source_file.startswith('.#'):
continue
if source_file.startswith('flycheck_'):
continue
source_str = ''
with open(source_file, 'r', encoding='utf-8') as file_source:
source_str = file_source.read()
if not source_str:
continue
for fname in fnames:
if fname + '(' not in source_str:
continue
names_list = source_str.split(fname + '(')
for index in range(1, len(names_list)):
param_var_name = None
if '"' in names_list[index]:
param_var_name = names_list[index].split('"')[1]
elif "'" in names_list[index]:
param_var_name = names_list[index].split("'")[1]
if not param_var_name:
continue
if ' ' in param_var_name:
continue
if '.' in param_var_name:
continue
if '/' in param_var_name:
continue
if '__' in param_var_name:
continue
if 'POST' in param_var_name:
continue
if param_var_name.isdigit():
continue
if '_' in param_var_name:
print(fname + ' in ' + source_file +
' should have camel case variable ' +
param_var_name)
assert False
break
def _test_source_contains_no_tabs():
print('test_source_contains_no_tabs')
for _, _, files in os.walk('.'):
for source_file in files:
if not source_file.endswith('.py'):
continue
if source_file.startswith('.#'):
continue
if source_file.startswith('flycheck_'):
continue
source_str = ''
with open(source_file, 'r', encoding='utf-8') as file_source:
source_str = file_source.read()
if not source_str:
continue
if '\t' in source_str:
print(source_file + ' contains tabs')
assert False
break
def _test_checkbox_names():
print('test_checkbox_names')
fnames = ['edit_text_field', 'edit_check_box', 'edit_text_area']
for _, _, files in os.walk('.'):
for source_file in files:
if not source_file.endswith('.py'):
continue
if source_file.startswith('.#'):
continue
if source_file.startswith('flycheck_'):
continue
source_str = ''
with open(source_file, 'r', encoding='utf-8') as file_source:
source_str = file_source.read()
if not source_str:
continue
for fname in fnames:
if fname + '(' not in source_str:
continue
names_list = source_str.split(fname + '(')
for index in range(1, len(names_list)):
if ')' not in names_list[index]:
continue
allparams = names_list[index].split(')')[0]
if ',' not in allparams:
continue
allparams_list = allparams.split(',')
if len(allparams_list) < 2:
continue
param_var_name = allparams_list[1].strip()
param_var_name = param_var_name.replace('"', '')
param_var_name = param_var_name.replace("'", '')
if ' ' in param_var_name:
continue
if '/' in param_var_name:
continue
if '_' in param_var_name:
print(fname + ' in ' + source_file +
' should have camel case variable ' +
param_var_name)
assert False
break
def _test_post_field_names(source_file: str, fieldnames: []):
print('test_post_field_Names')
fnames = []
for field in fieldnames:
fnames.append(field + '.get')
source_str = ''
with open(source_file, 'r', encoding='utf-8') as file_source:
source_str = file_source.read()
if not source_str:
return
for fname in fnames:
if fname + '(' not in source_str:
continue
names_list = source_str.split(fname + '(')
for index in range(1, len(names_list)):
if ')' not in names_list[index]:
continue
param_var_name = names_list[index].split(')')[0].strip()
if '"' not in param_var_name and \
"'" not in param_var_name:
continue
quote_str = '"'
if "'" in param_var_name:
quote_str = "'"
orig_param_var_name = fname + '(' + param_var_name + ')'
param_var_name = param_var_name.split(quote_str)[1]
if ' ' in param_var_name:
continue
if '/' in param_var_name:
continue
if '_' in param_var_name:
print(orig_param_var_name + ' in ' + source_file +
' should be camel case')
assert False
fnames = []
for field in fieldnames:
fnames.append(field + '[')
for fname in fnames:
if fname in source_str:
names_list = source_str.split(fname)
for index in range(1, len(names_list)):
if ']' not in names_list[index]:
continue
param_var_name = names_list[index].split(']')[0].strip()
if '"' not in param_var_name and \
"'" not in param_var_name:
continue
quote_str = '"'
if "'" in param_var_name:
quote_str = "'"
orig_param_var_name = fname.strip() + param_var_name + ']'
param_var_name = param_var_name.split(quote_str)[1]
if ' ' in param_var_name:
continue
if '/' in param_var_name:
continue
if '_' in param_var_name:
print(orig_param_var_name + ' in ' + source_file +
' should be camel case')
assert False
def _test_thread_functions():
print('test_thread_functions')
modules = {}
threads_called_in_modules = []
# get the source for each module
# Allow recursive walk
for _, _, files in os.walk('.'):
for source_file in files:
if not source_file.endswith('.py'):
continue
if source_file.startswith('.#'):
continue
if source_file.startswith('flycheck_'):
continue
mod_name = source_file.replace('.py', '')
if mod_name == 'threads':
# don't test the threads module itself
continue
modules[mod_name] = {
'functions': []
}
source_str = ''
with open(source_file, 'r', encoding='utf-8') as fp_src:
source_str = fp_src.read()
modules[mod_name]['source'] = source_str
if 'thread_with_trace(' in source_str:
threads_called_in_modules.append(mod_name)
with open(source_file, 'r', encoding='utf-8') as fp_src:
lines = fp_src.readlines()
modules[mod_name]['lines'] = lines
for mod_name in threads_called_in_modules:
thread_sections = \
modules[mod_name]['source'].split('thread_with_trace(')
ctr = 0
for thread_str in thread_sections:
if ctr == 0 or ',' not in thread_str:
ctr += 1
continue
thread_function_args = thread_str.split(',')
first_parameter = thread_function_args[0]
if 'target=' not in first_parameter:
ctr += 1
continue
thread_function_name = first_parameter.split('target=')[1].strip()
if not thread_function_name:
ctr += 1
continue
if thread_function_name.startswith('self.'):
thread_function_name = thread_function_name.split('self.')[1]
# is this function declared at the top of the module
# or defined within the module?
import_str = ' import ' + thread_function_name
def_str = 'def ' + thread_function_name + '('
if import_str not in modules[mod_name]['source']:
if def_str not in modules[mod_name]['source']:
print(mod_name + ' ' + first_parameter)
print(import_str + ' not in ' + mod_name)
assert False
if def_str in modules[mod_name]['source']:
defininition_module = mod_name
else:
# which module is the thread function defined within?
test_str = modules[mod_name]['source'].split(import_str)[0]
defininition_module = test_str.split('from ')[-1]
print('Thread function ' + thread_function_name +
' defined in ' + defininition_module)
# check the function arguments
second_parameter = thread_function_args[1]
if 'args=' not in second_parameter:
print('No args parameter in ' + thread_function_name +
' module ' + mod_name)
ctr += 1
continue
arg_ctr = 0
calling_function_args_list = []
for func_arg in thread_function_args:
if arg_ctr == 0:
arg_ctr += 1
continue
last_arg = False
if '(' in func_arg and '()' not in func_arg:
func_arg = func_arg.split('(')[1]
if func_arg.endswith(')') and '()' not in func_arg:
func_arg = func_arg.split(')')[0]
last_arg = True
func_arg = func_arg.strip()
calling_function_args_list.append(func_arg)
if last_arg:
break
arg_ctr += 1
# print(mod_name + ' ' + thread_function_name + ' ' +
# str(calling_function_args_list))
# get the function definition arguments
test_str = \
modules[defininition_module]['source'].split(def_str)[1]
test_str = test_str.split(')')[0]
definition_function_args_list = test_str.split(',')
if len(definition_function_args_list) > 0:
if definition_function_args_list[0] == 'self':
definition_function_args_list = \
definition_function_args_list[1:]
if len(definition_function_args_list) != \
len(calling_function_args_list):
print('Thread function ' + thread_function_name +
' has ' + str(len(definition_function_args_list)) +
' arguments, but ' +
str(len(calling_function_args_list)) +
' were given in module ' + mod_name)
print(str(definition_function_args_list))
print(str(calling_function_args_list))
assert False
ctr += 1
def _test_functions():
print('test_functions')
function = {}
function_properties = {}
modules = {}
mod_groups = {}
method_loc = []
for _, _, files in os.walk('.'):
for source_file in files:
# is this really a source file?
if not source_file.endswith('.py'):
continue
if source_file.startswith('.#'):
continue
if source_file.startswith('flycheck_'):
continue
# get the module name
mod_name = source_file.replace('.py', '')
modules[mod_name] = {
'functions': []
}
# load the module source
source_str = ''
with open(source_file, 'r', encoding='utf-8') as fp_src:
source_str = fp_src.read()
modules[mod_name]['source'] = source_str
# go through the source line by line
with open(source_file, 'r', encoding='utf-8') as fp_src:
lines = fp_src.readlines()
modules[mod_name]['lines'] = lines
line_count = 0
prev_line = 'start'
method_name = ''
for line in lines:
# what group is this module in?
if '__module_group__' in line:
if '=' in line:
group_name = line.split('=')[1].strip()
group_name = group_name.replace('"', '')
group_name = group_name.replace("'", '')
modules[mod_name]['group'] = group_name
if not mod_groups.get(group_name):
mod_groups[group_name] = [mod_name]
else:
if mod_name not in mod_groups[group_name]:
mod_groups[group_name].append(mod_name)
# reading function lines
if not line.strip().startswith('def '):
if line_count > 0:
line_count += 1
# add LOC count for this function
if len(prev_line.strip()) == 0 and \
len(line.strip()) == 0 and \
line_count > 2:
line_count -= 2
if line_count > 80:
loc_str = str(line_count) + ';' + method_name
if line_count < 1000:
loc_str = '0' + loc_str
if line_count < 100:
loc_str = '0' + loc_str
if line_count < 10:
loc_str = '0' + loc_str
if loc_str not in method_loc:
method_loc.append(loc_str)
line_count = 0
prev_line = line
continue
# reading function def
prev_line = line
line_count = 1
method_name = line.split('def ', 1)[1].split('(')[0]
# get list of arguments with spaces removed
method_args = \
source_str.split('def ' + method_name + '(')[1]
method_args = method_args.split(')')[0]
method_args = method_args.replace(' ', '').split(',')
if function.get(mod_name):
function[mod_name].append(method_name)
else:
function[mod_name] = [method_name]
if method_name not in modules[mod_name]['functions']:
modules[mod_name]['functions'].append(method_name)
# create an entry for this function
function_properties[method_name] = {
"args": method_args,
"module": mod_name,
"calledInModule": []
}
# LOC count for the last function
if line_count > 2:
line_count -= 2
if line_count > 80:
loc_str = str(line_count) + ';' + method_name
if line_count < 1000:
loc_str = '0' + loc_str
if line_count < 100:
loc_str = '0' + loc_str
if line_count < 10:
loc_str = '0' + loc_str
if loc_str not in method_loc:
method_loc.append(loc_str)
break
print('LOC counts:')
method_loc.sort()
for loc_str in method_loc:
print(loc_str.split(';')[0] + ' ' + loc_str.split(';')[1])
exclude_func_args = [
'pyjsonld'
]
exclude_funcs = [
'link',
'set',
'get'
]
bad_function_names = []
for name, properties in function_properties.items():
if '_' not in name:
if name.lower() != name:
bad_function_names.append(name)
else:
if name.startswith('_'):
name2 = name[1:]
if '_' not in name2:
if name2.lower() != name2:
bad_function_names.append(name)
if bad_function_names:
bad_function_names2 = \
sorted(bad_function_names, key=len, reverse=True)
function_names_str = ''
function_names_sh = ''
ctr = 0
for name in bad_function_names2:
if ctr > 0:
function_names_str += '\n' + name
else:
function_names_str = name
snake_case_name = convert_to_snake_case(name)
function_names_sh += \
'sed -i "s|' + name + '|' + snake_case_name + '|g" *.py\n'
ctr += 1
print(function_names_str + '\n')
with open('scripts/bad_function_names.sh', 'w+',
encoding='utf-8') as fn_file:
fn_file.write(function_names_sh)
assert False
# which modules is each function used within?
for mod_name, _ in modules.items():
print('Module: ' + mod_name + ' ✓')
for name, properties in function_properties.items():
line_ctr = 0
for line in modules[mod_name]['lines']:
line_str = line.strip()
if line_str.startswith('def '):
line_ctr += 1
continue
if line_str.startswith('class '):
line_ctr += 1
continue
# detect a call to this function
if name + '(' in line:
mod_list = \
function_properties[name]['calledInModule']
if mod_name not in mod_list:
mod_list.append(mod_name)
if mod_name in exclude_func_args:
line_ctr += 1
continue
if name in exclude_funcs:
line_ctr += 1
continue
# get the function call arguments
call_args = \
_get_function_call_args(name,
modules[mod_name]['lines'],
line_ctr)
# get the function def arguments
func_args = function_properties[name]['args']
# match the call arguments to the definition arguments
if not _function_args_match(call_args, func_args):
print('Call to function ' + name +
' does not match its arguments')
print('def args: ' +
str(len(function_properties[name]['args'])) +
'\n' + str(function_properties[name]['args']))
print('Call args: ' + str(len(call_args)) + '\n' +
str(call_args))
print('module ' + mod_name + ' line ' + str(line_ctr))
assert False
line_ctr += 1
# don't check these functions, because they are procedurally called
exclusions = [
'do_GET',
'do_POST',
'do_HEAD',
'do_PROPFIND',
'do_PUT',
'do_REPORT',
'do_DELETE',
'__run',
'_send_to_named_addresses',
'globaltrace',
'localtrace',
'kill',
'clone',
'unregister_rdf_parser',
'set_document_loader',
'has_property',
'has_value',
'add_value',
'get_values',
'remove_property',
'remove_value',
'normalize',
'get_document_loader',
'run_inbox_queue_watchdog',
'run_inbox_queue',
'run_import_following',
'run_import_following_watchdog',
'run_post_schedule',
'run_post_schedule_watchdog',
'str2bool',
'run_newswire_daemon',
'run_newswire_watchdog',
'run_federated_shares_watchdog',
'run_federated_shares_daemon',
'fitness_thread',
'thread_send_post',
'send_to_followers',
'expire_cache',
'get_mutuals_of_person',
'run_posts_queue',
'run_shares_expire',
'run_posts_watchdog',
'run_shares_expire_watchdog',
'get_this_weeks_events',
'get_availability',
'_test_threads_function',
'create_server_group',
'create_server_alice',
'create_server_bob',
'create_server_eve',
'e2e_eremove_device',
'setOrganizationScheme',
'fill_headers',
'_nothing',
'check_for_changed_actor'
]
exclude_imports = [
'link',
'start'
]
exclude_local = [
'pyjsonld',
'daemon',
'tests'
]
exclude_mods = [
'pyjsonld'
]
# check that functions are called somewhere
for name, properties in function_properties.items():
if name.startswith('__'):
if name.endswith('__'):
continue
if name in exclusions:
continue
if properties['module'] in exclude_mods:
continue
is_local_function = False
if not properties['calledInModule']:
print('function ' + name +
' in module ' + properties['module'] +
' is not called anywhere')
assert properties['calledInModule']
if len(properties['calledInModule']) == 1:
mod_name = properties['calledInModule'][0]
if mod_name not in exclude_local and \
mod_name == properties['module']:
is_local_function = True
if not name.startswith('_'):
print('Local function ' + name +
' in ' + mod_name + '.py does not begin with _')
assert False
if name not in exclude_imports:
for mod_name in properties['calledInModule']:
if mod_name == properties['module']:
continue
import_str = 'from ' + properties['module'] + ' import ' + name
if import_str not in modules[mod_name]['source']:
print(import_str + ' not found in ' + mod_name + '.py')
assert False
if not is_local_function:
if name.startswith('_'):
exclude_public = [
'pyjsonld',
'daemon',
'tests'
]
mod_name = properties['module']
if mod_name not in exclude_public:
print('Public function ' + name + ' in ' +
mod_name + '.py begins with _')
assert False
print('Function: ' + name + ' ✓')
print('Constructing function call graph')
module_colors = (
'red', 'green', 'yellow', 'orange', 'purple', 'cyan',
'darkgoldenrod3', 'darkolivegreen1', 'darkorange1',
'darkorchid1', 'darkseagreen', 'darkslategray4',
'deeppink1', 'deepskyblue1', 'dimgrey', 'gold1',
'goldenrod', 'burlywood2', 'bisque1', 'brown1',
'chartreuse2', 'cornsilk', 'darksalmon'
)
max_module_calls = 1
max_function_calls = 1
color_ctr = 0
for mod_name, _ in modules.items():
line_ctr = 0
modules[mod_name]['color'] = module_colors[color_ctr]
color_ctr += 1
if color_ctr >= len(module_colors):
color_ctr = 0
for line in modules[mod_name]['lines']:
if line.strip().startswith('def '):
name = line.split('def ')[1].split('(')[0]
calls_list = \
get_function_calls(name, modules[mod_name]['lines'],
line_ctr, function_properties)
function_properties[name]['calls'] = calls_list.copy()
if len(calls_list) > max_function_calls:
max_function_calls = len(calls_list)
# keep track of which module calls which other module
for fnc in calls_list:
mod_call = function_properties[fnc]['module']
if mod_call != mod_name:
if modules[mod_name].get('calls'):
if mod_call not in modules[mod_name]['calls']:
modules[mod_name]['calls'].append(mod_call)
if len(modules[mod_name]['calls']) > \
max_module_calls:
max_module_calls = \
len(modules[mod_name]['calls'])
else:
modules[mod_name]['calls'] = [mod_call]
line_ctr += 1
_diagram_groups(['Commandline Interface', 'ActivityPub'], ['utils'],
modules, mod_groups, max_module_calls)
_diagram_groups(['Commandline Interface', 'Core'], ['utils'],
modules, mod_groups, max_module_calls)
_diagram_groups(['Timeline', 'Core'], ['utils'],
modules, mod_groups, max_module_calls)
_diagram_groups(['Web Interface', 'Core'], ['utils'],
modules, mod_groups, max_module_calls)
_diagram_groups(['Web Interface Columns', 'Core'], ['utils'],
modules, mod_groups, max_module_calls)
_diagram_groups(['Core'], [],
modules, mod_groups, max_module_calls)
_diagram_groups(['ActivityPub'], [],
modules, mod_groups, max_module_calls)
_diagram_groups(['ActivityPub', 'Core'], ['utils'],
modules, mod_groups, max_module_calls)
_diagram_groups(['ActivityPub', 'Security'], ['utils'],
modules, mod_groups, max_module_calls)
_diagram_groups(['Core', 'Security'], ['utils'],
modules, mod_groups, max_module_calls)
_diagram_groups(['Timeline', 'Security'], ['utils'],
modules, mod_groups, max_module_calls)
_diagram_groups(['Web Interface', 'Accessibility'],
['utils', 'webapp_utils'],
modules, mod_groups, max_module_calls)
_diagram_groups(['Core', 'Accessibility'], ['utils'],
modules, mod_groups, max_module_calls)
def _test_links_within_post(base_dir: str) -> None:
print('test_links_within_post')
system_language = 'en'
languages_understood = [system_language]
nickname = 'test27636'
domain = 'rando.site'
port = 443
http_prefix = 'https'
content = 'This is a test post with links.\n\n' + \
'ftp://ftp.ncdc.noaa.gov/pub/data/ghcn/v4/\n\nhttps://libreserver.org'
save_to_file = False
client_to_server = False
comments_enabled = True
attach_image_filename = None
media_type = None
image_description = None
city = 'London, England'
test_in_reply_to = None
test_in_reply_to_atom_uri = None
test_subject = None
test_schedule_post = False
test_event_date = None
test_event_time = None
test_event_end_time = None
test_location = None
test_is_article = False
conversation_id = None
low_bandwidth = True
content_license_url = 'https://creativecommons.org/licenses/by-nc/4.0'
media_license_url = 'https://creativecommons.org/licenses/by-nc/4.0'
media_creator = 'Dr No'
translate = {}
buy_url = ''
video_transcript = ''
post_json_object = \
create_public_post(base_dir, nickname, domain, port, http_prefix,
content, save_to_file,
client_to_server, comments_enabled,
attach_image_filename, media_type,
image_description, video_transcript, city,
test_in_reply_to, test_in_reply_to_atom_uri,
test_subject, test_schedule_post,
test_event_date, test_event_time,
test_event_end_time, test_location,
test_is_article, system_language, conversation_id,
low_bandwidth, content_license_url,
media_license_url, media_creator,
languages_understood, translate, buy_url)
expected_str = \
'
\n'
result = markdown_to_html(markdown)
if result != expected:
print(result)
assert result == expected
markdown = \
'This is [a link](https://something.somewhere) to something.\n' + \
'And [something else](https://cat.pic).\n' + \
'Or .'
expected = \
'This is ' + \
'a link to something. \n' + \
'And ' + \
'something else. \n' + \
'Or .'
result = markdown_to_html(markdown)
if result != expected:
print(result)
assert result == expected
def _test_extract_text_fields_from_post():
print('test_extract_text_fields_in_post')
boundary = '--LYNX'
form_data = '--LYNX\r\nContent-Disposition: form-data; ' + \
'name="fieldName"\r\nContent-Type: text/plain; ' + \
'charset=utf-8\r\n\r\nThis is a lynx test\r\n' + \
'--LYNX\r\nContent-Disposition: ' + \
'form-data; name="submitYes"\r\nContent-Type: text/plain; ' + \
'charset=utf-8\r\n\r\nBUTTON\r\n--LYNX--\r\n'
debug = True
fields = extract_text_fields_in_post(None, boundary, debug, form_data)
print('fields: ' + str(fields))
assert fields
assert fields['fieldName'] == 'This is a lynx test'
assert fields['submitYes'] == 'BUTTON'
boundary = '-----------------------------116202748023898664511855843036'
form_data = '-----------------------------116202748023898664511855' + \
'843036\r\nContent-Disposition: form-data; name="submitPost"' + \
'\r\n\r\nSubmit\r\n-----------------------------116202748023' + \
'898664511855843036\r\nContent-Disposition: form-data; name=' + \
'"subject"\r\n\r\n\r\n-----------------------------116202748' + \
'023898664511855843036\r\nContent-Disposition: form-data; na' + \
'me="message"\r\n\r\nThis is a ; test\r\n-------------------' + \
'----------116202748023898664511855843036\r\nContent-Disposi' + \
'tion: form-data; name="commentsEnabled"\r\n\r\non\r\n------' + \
'-----------------------116202748023898664511855843036\r\nCo' + \
'ntent-Disposition: form-data; name="eventDate"\r\n\r\n\r\n' + \
'-----------------------------116202748023898664511855843036' + \
'\r\nContent-Disposition: form-data; name="eventTime"\r\n\r' + \
'\n\r\n-----------------------------116202748023898664511855' + \
'843036\r\nContent-Disposition: form-data; name="location"' + \
'\r\n\r\n\r\n-----------------------------116202748023898664' + \
'511855843036\r\nContent-Disposition: form-data; name=' + \
'"imageDescription"\r\n\r\n\r\n-----------------------------' + \
'116202748023898664511855843036\r\nContent-Disposition: ' + \
'form-data; name="attachpic"; filename=""\r\nContent-Type: ' + \
'application/octet-stream\r\n\r\n\r\n----------------------' + \
'-------116202748023898664511855843036--\r\n' + \
'Content-Disposition: form-data; name="importBlocks"; ' + \
'filename="wildebeest_suspend.csv"\r\nContent-Type: ' + \
'text/csv\r\n\r\n#domain,#severity,#reject_media,#reject_reports,' + \
'#public_comment,#obfuscate\nbgp.social,suspend,false,false,' + \
'"Wildebeest",false\ncesko.social,suspend,false,false,' + \
'"Wildebeest",false\ncloudflare.social,suspend,false,false,' + \
'"Wildebeest",false\ndogfood.social,suspend,false,false,' + \
'"Wildebeest",false\ndomo.cafe,suspend,false,false,"Wildebeest",' + \
'false\nemaw.social,suspend,false,false\n\r\n ' + \
'-----------------------------116202748023898664511855843036--\r\n'
debug = False
fields = extract_text_fields_in_post(None, boundary, debug, form_data)
assert fields['submitPost'] == 'Submit'
assert fields['subject'] == ''
assert fields['commentsEnabled'] == 'on'
assert fields['eventDate'] == ''
assert fields['eventTime'] == ''
assert fields['location'] == ''
assert fields['imageDescription'] == ''
assert fields['message'] == 'This is a ; test'
if not fields['importBlocks'][1:].startswith('#domain,#severity,'):
print(fields['importBlocks'])
assert fields['importBlocks'][1:].startswith('#domain,#severity,')
def _test_speaker_replace_link():
http_prefix = 'https'
nickname = 'mynick'
domain = 'mydomain'
domain_full = domain
print('testSpeakerReplaceLinks')
text = 'The Tor Project: For Snowflake volunteers: If you use ' + \
'Firefox, Brave, or Chrome, our Snowflake extension turns ' + \
'your browser into a proxy that connects Tor users in ' + \
'censored regions to the Tor network. Note: you should ' + \
'not run more than one snowflake in the same ' + \
'network.https://support.torproject.org/censorship/' + \
'how-to-help-running-snowflake/'
detected_links = []
result = \
speaker_replace_links(http_prefix, nickname, domain, domain_full,
text, {'Linked': 'Web link'}, detected_links)
assert len(detected_links) == 1
assert detected_links[0] == \
'https://support.torproject.org/censorship/' + \
'how-to-help-running-snowflake/'
assert 'Web link support.torproject.org' in result
remote_link = 'https://somedomain/tags/sometag'
text = 'Test with a hashtag ' + remote_link + ' link'
detected_links = []
result = \
speaker_replace_links(http_prefix, nickname, domain, domain_full,
text, {'Linked': 'Web link'}, detected_links)
assert len(detected_links) == 1
local_link = \
'https://' + domain_full + '/users/' + nickname + \
'?remotetag=' + remote_link.replace('/', '--')
assert detected_links[0] == local_link
assert 'Web link somedomain' in result
def _test_camel_case_split():
print('test_camel_case_split')
test_str = 'ThisIsCamelCase'
assert camel_case_split(test_str) == 'This Is Camel Case'
test_str = 'Notcamelcase test'
assert camel_case_split(test_str) == 'Notcamelcase test'
def _test_emoji_images():
print('test_emoji_images')
emoji_filename = 'emoji/default_emoji.json'
assert os.path.isfile(emoji_filename)
emoji_json = load_json(emoji_filename)
assert emoji_json
for emoji_name, emoji_image in emoji_json.items():
emoji_image_filename = 'emoji/' + emoji_image + '.png'
if not os.path.isfile(emoji_image_filename):
print('Missing emoji image ' + emoji_name + ' ' +
emoji_image + '.png')
assert os.path.isfile(emoji_image_filename)
def _test_extract_pgp_public_key():
print('testExtractPGPPublicKey')
pub_key = \
'-----BEGIN PGP PUBLIC KEY BLOCK-----\n\n' + \
'mDMEWZBueBYJKwYBBAHaRw8BAQdAKx1t6wL0RTuU6/' + \
'IBjngMbVJJ3Wg/3UW73/PV\n' + \
'I47xKTS0IUJvYiBNb3R0cmFtIDxib2JAZnJlZWRvb' + \
'WJvbmUubmV0PoiQBBMWCAA4\n' + \
'FiEEmruCwAq/OfgmgEh9zCU2GR+nwz8FAlmQbngCG' + \
'wMFCwkIBwMFFQoJCAsFFgID\n' + \
'AQACHgECF4AACgkQzCU2GR+nwz/9sAD/YgsHnVszH' + \
'Nz1zlVc5EgY1ByDupiJpHj0\n' + \
'XsLYk3AbNRgBALn45RqgD4eWHpmOriH09H5Rc5V9i' + \
'N4+OiGUn2AzJ6oHuDgEWZBu\n' + \
'eBIKKwYBBAGXVQEFAQEHQPRBG2ZQJce475S3e0Dxe' + \
'b0Fz5WdEu2q3GYLo4QG+4Ry\n' + \
'AwEIB4h4BBgWCAAgFiEEmruCwAq/OfgmgEh9zCU2G' + \
'R+nwz8FAlmQbngCGwwACgkQ\n' + \
'zCU2GR+nwz+OswD+JOoyBku9FzuWoVoOevU2HH+bP' + \
'OMDgY2OLnST9ZSyHkMBAMcK\n' + \
'fnaZ2Wi050483Sj2RmQRpb99Dod7rVZTDtCqXk0J\n' + \
'=gv5G\n' + \
'-----END PGP PUBLIC KEY BLOCK-----'
test_str = "Some introduction\n\n" + pub_key + "\n\nSome message."
assert contains_pgp_public_key(test_str)
assert not contains_pgp_public_key('String without a pgp key')
result = extract_pgp_public_key(test_str)
assert result
assert result == pub_key
def test_update_actor(base_dir: str):
print('Testing update of actor properties')
global TEST_SERVER_ALICE_RUNNING
TEST_SERVER_ALICE_RUNNING = False
http_prefix = 'http'
proxy_type = None
federation_list = []
if os.path.isdir(base_dir + '/.tests'):
shutil.rmtree(base_dir + '/.tests',
ignore_errors=False, onerror=None)
os.mkdir(base_dir + '/.tests')
# create the server
alice_dir = base_dir + '/.tests/alice'
alice_domain = '127.0.0.11'
alice_port = 61792
alice_send_threads = []
bob_address = '127.0.0.84:6384'
global THR_ALICE
if THR_ALICE:
while THR_ALICE.is_alive():
THR_ALICE.stop()
time.sleep(1)
THR_ALICE.kill()
THR_ALICE = \
thread_with_trace(target=create_server_alice,
args=(alice_dir, alice_domain,
alice_port, bob_address,
federation_list, False, False,
alice_send_threads),
daemon=True)
THR_ALICE.start()
assert THR_ALICE.is_alive() is True
# wait for server to be running
ctr = 0
while not TEST_SERVER_ALICE_RUNNING:
time.sleep(1)
ctr += 1
if ctr > 60:
break
print('Alice online: ' + str(TEST_SERVER_ALICE_RUNNING))
print('\n\n*******************************************************')
print('Alice updates her PGP key')
session_alice = create_session(proxy_type)
cached_webfingers = {}
person_cache = {}
password = 'alicepass'
outbox_path = alice_dir + '/accounts/alice@' + alice_domain + '/outbox'
actor_filename = \
alice_dir + '/accounts/' + 'alice@' + alice_domain + '.json'
assert os.path.isfile(actor_filename)
assert len([name for name in os.listdir(outbox_path)
if os.path.isfile(os.path.join(outbox_path, name))]) == 0
pub_key = \
'-----BEGIN PGP PUBLIC KEY BLOCK-----\n\n' + \
'mDMEWZBueBYJKwYBBAHaRw8BAQdAKx1t6wL0RTuU6/' + \
'IBjngMbVJJ3Wg/3UW73/PV\n' + \
'I47xKTS0IUJvYiBNb3R0cmFtIDxib2JAZnJlZWRvb' + \
'WJvbmUubmV0PoiQBBMWCAA4\n' + \
'FiEEmruCwAq/OfgmgEh9zCU2GR+nwz8FAlmQbngCG' + \
'wMFCwkIBwMFFQoJCAsFFgID\n' + \
'AQACHgECF4AACgkQzCU2GR+nwz/9sAD/YgsHnVszH' + \
'Nz1zlVc5EgY1ByDupiJpHj0\n' + \
'XsLYk3AbNRgBALn45RqgD4eWHpmOriH09H5Rc5V9i' + \
'N4+OiGUn2AzJ6oHuDgEWZBu\n' + \
'eBIKKwYBBAGXVQEFAQEHQPRBG2ZQJce475S3e0Dxe' + \
'b0Fz5WdEu2q3GYLo4QG+4Ry\n' + \
'AwEIB4h4BBgWCAAgFiEEmruCwAq/OfgmgEh9zCU2G' + \
'R+nwz8FAlmQbngCGwwACgkQ\n' + \
'zCU2GR+nwz+OswD+JOoyBku9FzuWoVoOevU2HH+bP' + \
'OMDgY2OLnST9ZSyHkMBAMcK\n' + \
'fnaZ2Wi050483Sj2RmQRpb99Dod7rVZTDtCqXk0J\n' + \
'=gv5G\n' + \
'-----END PGP PUBLIC KEY BLOCK-----'
signing_priv_key_pem = None
actor_update = \
pgp_public_key_upload(alice_dir, session_alice,
'alice', password,
alice_domain, alice_port,
http_prefix,
cached_webfingers, person_cache,
True, pub_key, signing_priv_key_pem)
print('actor update result: ' + str(actor_update))
assert actor_update
# load alice actor
print('Loading actor: ' + actor_filename)
actor_json = load_json(actor_filename)
assert actor_json
if len(actor_json['attachment']) == 0:
print("actor_json['attachment'] has no contents")
assert len(actor_json['attachment']) > 0
property_found = False
for property_value in actor_json['attachment']:
if property_value['name'] == 'PGP':
print('PGP property set within attachment')
assert pub_key in property_value['value']
property_found = True
assert property_found
# stop the server
THR_ALICE.kill()
THR_ALICE.join()
assert THR_ALICE.is_alive() is False
os.chdir(base_dir)
if os.path.isdir(base_dir + '/.tests'):
shutil.rmtree(base_dir + '/.tests', ignore_errors=False, onerror=None)
def _test_remove_interactions() -> None:
print('test_remove_post_interactions')
post_json_object = {
"type": "Create",
"object": {
"to": ["#Public"],
"likes": {
"items": ["a", "b", "c"]
},
"replies": {
"replyStuff": ["a", "b", "c"]
},
"shares": {
"sharesStuff": ["a", "b", "c"]
},
"bookmarks": {
"bookmarksStuff": ["a", "b", "c"]
},
"ignores": {
"ignoresStuff": ["a", "b", "c"]
}
}
}
remove_post_interactions(post_json_object, True)
assert post_json_object['object']['likes']['items'] == []
assert post_json_object['object']['replies'] == {}
assert post_json_object['object']['shares'] == {}
assert post_json_object['object']['bookmarks'] == {}
assert post_json_object['object']['ignores'] == {}
post_json_object['object']['to'] = ["some private address"]
assert not remove_post_interactions(post_json_object, False)
def _test_spoofed_geolocation() -> None:
print('test_spoof_geolocation')
nogo_line = \
'NEW YORK, USA: 73.951W,40.879, 73.974W,40.83, ' + \
'74.029W,40.756, 74.038W,40.713, 74.056W,40.713, ' + \
'74.127W,40.647, 74.038W,40.629, 73.995W,40.667, ' + \
'74.014W,40.676, 73.994W,40.702, 73.967W,40.699, ' + \
'73.958W,40.729, 73.956W,40.745, 73.918W,40.781, ' + \
'73.937W,40.793, 73.946W,40.782, 73.977W,40.738, ' + \
'73.98W,40.713, 74.012W,40.705, 74.006W,40.752, ' + \
'73.955W,40.824'
polygon = parse_nogo_string(nogo_line)
assert len(polygon) > 0
assert polygon[0][1] == -73.951
assert polygon[0][0] == 40.879
cities_list = [
'NEW YORK, USA:40.7127281:W74.0060152:784',
'LOS ANGELES, USA:34.0536909:W118.242766:1214',
'SAN FRANCISCO, USA:37.74594738515095:W122.44299445520019:121',
'HOUSTON, USA:29.6072:W95.1586:1553',
'MANCHESTER, ENGLAND:53.4794892:W2.2451148:1276',
'BERLIN, GERMANY:52.5170365:13.3888599:891',
'ANKARA, TURKEY:39.93:32.85:24521',
'LONDON, ENGLAND:51.5073219:W0.1276474:1738',
'SEATTLE, USA:47.59840153253106:W122.31143714060059:217'
]
test_square = [
[[0.03, 0.01], [0.02, 10], [10.01, 10.02], [10.03, 0.02]]
]
assert point_in_nogo(test_square, 5, 5)
assert point_in_nogo(test_square, 2, 3)
assert not point_in_nogo(test_square, 20, 5)
assert not point_in_nogo(test_square, 11, 6)
assert not point_in_nogo(test_square, 5, -5)
assert not point_in_nogo(test_square, 5, 11)
assert not point_in_nogo(test_square, -5, -5)
assert not point_in_nogo(test_square, -5, 5)
nogo_list = []
curr_time = datetime.datetime.utcnow()
decoy_seed = 7634681
city_radius = 0.1
coords = spoof_geolocation('', 'los angeles', curr_time,
decoy_seed, cities_list, nogo_list)
assert coords[0] >= 34.0536909 - city_radius
assert coords[0] <= 34.0536909 + city_radius
assert coords[1] >= 118.242766 - city_radius
assert coords[1] <= 118.242766 + city_radius
assert coords[2] == 'N'
assert coords[3] == 'W'
assert len(coords[4]) > 4
assert len(coords[5]) > 4
assert coords[6] > 0
nogo_list = []
coords = spoof_geolocation('', 'unknown', curr_time,
decoy_seed, cities_list, nogo_list)
assert coords[0] >= 51.8744 - city_radius
assert coords[0] <= 51.8744 + city_radius
assert coords[1] >= 0.368333 - city_radius
assert coords[1] <= 0.368333 + city_radius
assert coords[2] == 'N'
assert coords[3] == 'W'
assert len(coords[4]) == 0
assert len(coords[5]) == 0
assert coords[6] == 0
kml_str = '\n'
kml_str += '\n'
kml_str += '\n'
nogo_line2 = \
'NEW YORK, USA: 74.115W,40.663, 74.065W,40.602, ' + \
'74.118W,40.555, 74.047W,40.516, 73.882W,40.547, ' + \
'73.909W,40.618, 73.978W,40.579, 74.009W,40.602, ' + \
'74.033W,40.61, 74.039W,40.623, 74.032W,40.641, ' + \
'73.996W,40.665'
polygon2 = parse_nogo_string(nogo_line2)
nogo_list = [polygon, polygon2]
for i in range(1000):
day_number = randint(10, 30)
hour = randint(1, 23)
hour_str = str(hour)
if hour < 10:
hour_str = '0' + hour_str
date_time_str = "2021-05-" + str(day_number) + " " + hour_str + ":14"
curr_time = datetime.datetime.strptime(date_time_str, "%Y-%m-%d %H:%M")
coords = spoof_geolocation('', 'new york, usa', curr_time,
decoy_seed, cities_list, nogo_list)
longitude = coords[1]
if coords[3] == 'W':
longitude = -coords[1]
kml_str += '\n'
kml_str += ' ' + str(i) + '\n'
kml_str += ' \n'
kml_str += ' ' + str(longitude) + ',' + \
str(coords[0]) + ',0\n'
kml_str += ' \n'
kml_str += '\n'
nogo_line = \
'LONDON, ENGLAND: 0.23888E,51.459, 0.1216E,51.5, ' + \
'0.016E,51.479, 0.097W,51.502, 0.126W,51.482, ' + \
'0.196W,51.457, 0.292W,51.465, 0.309W,51.49, ' + \
'0.226W,51.495, 0.198W,51.47, 0.174W,51.488, ' + \
'0.136W,51.489, 0.1189W,51.515, 0.038E,51.513, ' + \
'0.0692E,51.51, 0.12833E,51.526, 0.3289E,51.475'
polygon = parse_nogo_string(nogo_line)
nogo_line2 = \
'LONDON, ENGLAND: 0.054W,51.535, 0.044W,51.53, ' + \
'0.008W,51.55, 0.0429W,51.57, 0.038W,51.6, ' + \
'0.0209W,51.603, 0.032W,51.613, 0.00191E,51.66, ' + \
'0.024W,51.666, 0.0313W,51.659, 0.0639W,51.579, ' + \
'0.059W,51.568, 0.0329W,51.552'
polygon2 = parse_nogo_string(nogo_line2)
nogo_list = [polygon, polygon2]
for i in range(1000):
day_number = randint(10, 30)
hour = randint(1, 23)
hour_str = str(hour)
if hour < 10:
hour_str = '0' + hour_str
date_time_str = "2021-05-" + str(day_number) + " " + hour_str + ":14"
curr_time = datetime.datetime.strptime(date_time_str, "%Y-%m-%d %H:%M")
coords = spoof_geolocation('', 'london, england', curr_time,
decoy_seed, cities_list, nogo_list)
longitude = coords[1]
if coords[3] == 'W':
longitude = -coords[1]
kml_str += '\n'
kml_str += ' ' + str(i) + '\n'
kml_str += ' \n'
kml_str += ' ' + str(longitude) + ',' + \
str(coords[0]) + ',0\n'
kml_str += ' \n'
kml_str += '\n'
nogo_line = \
'SAN FRANCISCO, USA: 121.988W,37.408, 121.924W,37.452, ' + \
'121.951W,37.498, 121.992W,37.505, 122.056W,37.54, ' + \
'122.077W,37.578, 122.098W,37.618, 122.131W,37.637, ' + \
'122.189W,37.706, 122.227W,37.775, 122.279W,37.798, ' + \
'122.315W,37.802, 122.291W,37.832, 122.309W,37.902, ' + \
'122.382W,37.915, 122.368W,37.927, 122.514W,37.882, ' + \
'122.473W,37.83, 122.481W,37.788, 122.394W,37.796, ' + \
'122.384W,37.729, 122.4W,37.688, 122.382W,37.654, ' + \
'122.406W,37.637, 122.392W,37.612, 122.356W,37.586, ' + \
'122.332W,37.586, 122.275W,37.529, 122.228W,37.488, ' + \
'122.181W,37.482, 122.134W,37.48, 122.128W,37.471, ' + \
'122.122W,37.448, 122.095W,37.428, 122.07W,37.413, ' + \
'122.036W,37.402, 122.035W,37.421'
polygon = parse_nogo_string(nogo_line)
nogo_line2 = \
'SAN FRANCISCO, USA: 122.446W,37.794, 122.511W,37.778, ' + \
'122.51W,37.771, 122.454W,37.775, 122.452W,37.766, ' + \
'122.510W,37.763, 122.506W,37.735, 122.498W,37.733, ' + \
'122.496W,37.729, 122.491W,37.729, 122.475W,37.73, ' + \
'122.474W,37.72, 122.484W,37.72, 122.485W,37.703, ' + \
'122.495W,37.702, 122.493W,37.679, 122.486W,37.667, ' + \
'122.492W,37.664, 122.493W,37.629, 122.456W,37.625, ' + \
'122.450W,37.617, 122.455W,37.621, 122.41W,37.586, ' + \
'122.383W,37.561, 122.335W,37.509, 122.655W,37.48, ' + \
'122.67W,37.9, 122.272W,37.93, 122.294W,37.801, ' + \
'122.448W,37.804'
polygon2 = parse_nogo_string(nogo_line2)
nogo_list = [polygon, polygon2]
for i in range(1000):
day_number = randint(10, 30)
hour = randint(1, 23)
hour_str = str(hour)
if hour < 10:
hour_str = '0' + hour_str
date_time_str = "2021-05-" + str(day_number) + " " + hour_str + ":14"
curr_time = datetime.datetime.strptime(date_time_str, "%Y-%m-%d %H:%M")
coords = spoof_geolocation('', 'SAN FRANCISCO, USA', curr_time,
decoy_seed, cities_list, nogo_list)
longitude = coords[1]
if coords[3] == 'W':
longitude = -coords[1]
kml_str += '\n'
kml_str += ' ' + str(i) + '\n'
kml_str += ' \n'
kml_str += ' ' + str(longitude) + ',' + \
str(coords[0]) + ',0\n'
kml_str += ' \n'
kml_str += '\n'
nogo_line = \
'SEATTLE, USA: 122.247W,47.918, 122.39W,47.802, ' + \
'122.389W,47.769, 122.377W,47.758, 122.371W,47.726, ' + \
'122.379W,47.706, 122.4W,47.696, 122.405W,47.673, ' + \
'122.416W,47.65, 122.414W,47.642, 122.391W,47.632, ' + \
'122.373W,47.633, 122.336W,47.602, 122.288W,47.501, ' + \
'122.299W,47.503, 122.386W,47.592, 122.412W,47.574, ' + \
'122.394W,47.549, 122.388W,47.507, 122.35W,47.481, ' + \
'122.365W,47.459, 122.33W,47.406, 122.323W,47.392, ' + \
'122.321W,47.346, 122.441W,47.302, 122.696W,47.085, ' + \
'122.926W,47.066, 122.929W,48.383'
polygon = parse_nogo_string(nogo_line)
nogo_line2 = \
'SEATTLE, USA: 122.267W,47.758, 122.29W,47.471, ' + \
'122.272W,47.693, 122.256W,47.672, 122.278W,47.652, ' + \
'122.29W,47.583, 122.262W,47.548, 122.265W,47.52, ' + \
'122.218W,47.498, 122.194W,47.501, 122.193W,47.55, ' + \
'122.173W,47.58, 122.22W,47.617, 122.238W,47.617, ' + \
'122.239W,47.637, 122.2W,47.644, 122.207W,47.703, ' + \
'122.22W,47.705, 122.231W,47.699, 122.255W,47.751'
polygon2 = parse_nogo_string(nogo_line2)
nogo_line3 = \
'SEATTLE, USA: 122.347W,47.675, 122.344W,47.681, ' + \
'122.337W,47.685, 122.324W,47.679, 122.331W,47.677, ' + \
'122.34W,47.669, 122.34W,47.664, 122.348W,47.665'
polygon3 = parse_nogo_string(nogo_line3)
nogo_line4 = \
'SEATTLE, USA: 122.423W,47.669, 122.345W,47.641, ' + \
'122.34W,47.625, 122.327W,47.626, 122.274W,47.64, ' + \
'122.268W,47.654, 122.327W,47.654, 122.336W,47.647, ' + \
'122.429W,47.684'
polygon4 = parse_nogo_string(nogo_line4)
nogo_list = [polygon, polygon2, polygon3, polygon4]
for i in range(1000):
day_number = randint(10, 30)
hour = randint(1, 23)
hour_str = str(hour)
if hour < 10:
hour_str = '0' + hour_str
date_time_str = "2021-05-" + str(day_number) + " " + hour_str + ":14"
curr_time = datetime.datetime.strptime(date_time_str, "%Y-%m-%d %H:%M")
coords = spoof_geolocation('', 'SEATTLE, USA', curr_time,
decoy_seed, cities_list, nogo_list)
longitude = coords[1]
if coords[3] == 'W':
longitude = -coords[1]
kml_str += '\n'
kml_str += ' ' + str(i) + '\n'
kml_str += ' \n'
kml_str += ' ' + str(longitude) + ',' + \
str(coords[0]) + ',0\n'
kml_str += ' \n'
kml_str += '\n'
kml_str += '\n'
kml_str += ''
with open('unittest_decoy.kml', 'w+', encoding='utf-8') as kmlfile:
kmlfile.write(kml_str)
def _test_skills() -> None:
print('test_skills')
actor_json = {
'hasOccupation': [
{
'@type': 'Occupation',
'name': "Sysop",
"occupationLocation": {
"@type": "City",
"name": "Fediverse"
},
'skills': []
}
]
}
skills_dict = {
'bakery': 40,
'gardening': 70
}
set_skills_from_dict(actor_json, skills_dict)
assert actor_has_skill(actor_json, 'bakery')
assert actor_has_skill(actor_json, 'gardening')
assert actor_skill_value(actor_json, 'bakery') == 40
assert actor_skill_value(actor_json, 'gardening') == 70
def _test_roles() -> None:
print('test_roles')
actor_json = {
'hasOccupation': [
{
'@type': 'Occupation',
'name': "Sysop",
'occupationLocation': {
'@type': 'City',
'name': 'Fediverse'
},
'skills': []
}
]
}
test_roles_list = ["admin", "moderator"]
actor_roles_from_list(actor_json, test_roles_list)
assert actor_has_role(actor_json, "admin")
assert actor_has_role(actor_json, "moderator")
assert not actor_has_role(actor_json, "editor")
assert not actor_has_role(actor_json, "counselor")
assert not actor_has_role(actor_json, "artist")
def _test_useragent_domain() -> None:
print('test_user_agent_domain')
user_agent = \
'http.rb/4.4.1 (Mastodon/9.10.11; +https://mastodon.something/)'
agent_domain = user_agent_domain(user_agent, False)
if agent_domain != 'mastodon.something':
print(agent_domain)
assert agent_domain == 'mastodon.something'
user_agent = \
'Mozilla/70.0 (X11; Linux x86_64; rv:1.0) Gecko/20450101 Firefox/1.0'
assert user_agent_domain(user_agent, False) is None
def _test_switch_word(base_dir: str) -> None:
print('test_switch_words')
rules = [
"rock -> hamster",
"orange -> lemon"
]
nickname = 'testuser'
domain = 'testdomain.com'
content = 'This is a test'
result = switch_words(base_dir, nickname, domain, content, rules)
assert result == content
content = 'This is orange test'
result = switch_words(base_dir, nickname, domain, content, rules)
assert result == 'This is lemon test'
content = 'This is a test rock'
result = switch_words(base_dir, nickname, domain, content, rules)
assert result == 'This is a test hamster'
def _test_word_lengths_limit() -> None:
print('test_limit_word_lengths')
max_word_length = 13
text = "This is a test"
result = limit_word_lengths(text, max_word_length)
assert result == text
text = "This is an exceptionallylongword test"
result = limit_word_lengths(text, max_word_length)
assert result == "This is an exceptionally test"
def _test_limit_repeted_words() -> None:
print('test_limit_repeated_words')
text = \
"This is a preamble.\n\n" + \
"Same Same Same Same Same Same Same Same Same Same " + \
"Same Same Same Same Same Same Same Same Same Same " + \
"Same Same Same Same Same Same Same Same Same Same " + \
"Same Same Same Same Same Same Same Same Same Same " + \
"Same Same Same Same Same Same Same Same Same Same " + \
"Same Same Same Same Same Same Same Same Same Same " + \
"Same Same Same Same Same Same Same Same Same Same\n\n" + \
"Some other text."
expected = \
"This is a preamble.\n\n" + \
"Same Same Same Same Same Same\n\n" + \
"Some other text."
result = limit_repeated_words(text, 6)
assert result == expected
text = \
"This is other preamble.\n\n" + \
"Same Same Same Same Same Same Same Same Same Same " + \
"Same Same Same Same Same Same Same Same Same Same " + \
"Same Same Same Same Same Same Same Same Same Same " + \
"Same Same Same Same Same Same Same Same Same Same " + \
"Same Same Same Same Same Same Same Same Same Same " + \
"Same Same Same Same Same Same Same Same Same Same " + \
"Same Same Same Same Same Same Same Same Same Same " + \
"Same Same Same Same Same Same Same Same Same Same " + \
"Same Same Same Same Same Same Same Same Same Same"
expected = \
"This is other preamble.\n\n" + \
"Same Same Same Same Same Same"
result = limit_repeated_words(text, 6)
assert result == expected
def _test_set_actor_language():
print('test_set_actor_languages')
actor_json = {
"attachment": []
}
set_actor_languages(actor_json, 'es, fr, en')
assert len(actor_json['attachment']) == 1
assert actor_json['attachment'][0]['name'] == 'Languages'
assert actor_json['attachment'][0]['type'] == 'PropertyValue'
assert isinstance(actor_json['attachment'][0]['value'], str)
assert ',' in actor_json['attachment'][0]['value']
lang_list = get_actor_languages_list(actor_json)
assert 'en' in lang_list
assert 'fr' in lang_list
assert 'es' in lang_list
languages_str = get_actor_languages(actor_json)
assert languages_str == 'en / es / fr'
def _test_get_links_from_content():
print('test_get_links_from_content')
content = 'This text has no links'
links = get_links_from_content(content)
assert not links
link1 = 'https://somewebsite.net'
link2 = 'http://somewhere.or.other'
content = \
'This is @linked. ' + \
'And another.'
links = get_links_from_content(content)
assert len(links.items()) == 2
assert links.get('@linked')
assert links['@linked'] == link1
assert links.get('another')
assert links['another'] == link2
content_plain = '
' + remove_html(content) + '
'
assert '>@linked' not in content_plain
content = add_links_to_content(content_plain, links)
assert '>@linked' in content
def _test_authorized_shared_items():
print('test_authorize_shared_items')
shared_items_fed_domains = \
['dog.domain', 'cat.domain', 'birb.domain']
tokens_json = \
generate_shared_item_federation_tokens(shared_items_fed_domains, None)
tokens_json = \
create_shared_item_federation_token(None, 'cat.domain',
False, tokens_json)
assert tokens_json
assert not tokens_json.get('dog.domain')
assert tokens_json.get('cat.domain')
assert not tokens_json.get('birb.domain')
assert len(tokens_json['dog.domain']) == 0
assert len(tokens_json['cat.domain']) >= 64
assert len(tokens_json['birb.domain']) == 0
assert not authorize_shared_items(shared_items_fed_domains, None,
'birb.domain',
'cat.domain', 'M' * 86,
False, tokens_json)
assert authorize_shared_items(shared_items_fed_domains, None,
'birb.domain',
'cat.domain', tokens_json['cat.domain'],
False, tokens_json)
tokens_json = \
update_shared_item_federation_token(None,
'dog.domain', 'testToken',
True, tokens_json)
assert tokens_json['dog.domain'] == 'testToken'
# the shared item federation list changes
shared_items_federated_domains = \
['possum.domain', 'cat.domain', 'birb.domain']
tokens_json = \
merge_shared_item_tokens(None, '',
shared_items_federated_domains,
tokens_json)
assert 'dog.domain' not in tokens_json
assert 'cat.domain' in tokens_json
assert len(tokens_json['cat.domain']) >= 64
assert 'birb.domain' in tokens_json
assert 'possum.domain' in tokens_json
assert len(tokens_json['birb.domain']) == 0
assert len(tokens_json['possum.domain']) == 0
def _test_date_conversions() -> None:
print('test_date_conversions')
date_str = "2021-05-16T14:37:41Z"
date_sec = date_string_to_seconds(date_str)
date_str2 = date_seconds_to_string(date_sec)
assert date_str == date_str2
def _test_valid_password():
print('test_valid_password')
assert not valid_password('123')
assert not valid_password('')
assert valid_password('パスワード12345')
assert valid_password('测试密码12345')
assert valid_password('A!bc:defg1/234?56')
def _test_get_price_from_string() -> None:
print('test_get_price_from_string')
price, curr = get_price_from_string("5.23")
assert price == "5.23"
assert curr == "EUR"
price, curr = get_price_from_string("£7.36")
assert price == "7.36"
assert curr == "GBP"
price, curr = get_price_from_string("$10.63")
assert price == "10.63"
assert curr == "USD"
def _translate_ontology(base_dir: str) -> None:
print('test_translate_ontology')
return
ontology_types = get_category_types(base_dir)
url = 'https://translate.astian.org'
api_key = None
lt_lang_list = libretranslate_languages(url, api_key)
languages_str = get_supported_languages(base_dir)
assert languages_str
for otype in ontology_types:
changed = False
filename = base_dir + '/ontology/' + otype + 'Types.json'
if not os.path.isfile(filename):
continue
ontology_json = load_json(filename)
if not ontology_json:
continue
index = -1
for item in ontology_json['@graph']:
index += 1
if "rdfs:label" not in item:
continue
english_str = None
languages_found = []
for label in item["rdfs:label"]:
if '@language' not in label:
continue
languages_found.append(label['@language'])
if '@value' not in label:
continue
if label['@language'] == 'en':
english_str = label['@value']
if not english_str:
continue
for lang in languages_str:
if lang not in languages_found:
translated_str = None
if url and lang in lt_lang_list:
translated_str = \
libretranslate(url, english_str,
'en', lang, api_key)
if not translated_str:
translated_str = english_str
else:
translated_str = translated_str.replace('
', '')
translated_str = translated_str.replace('
', '')
ontology_json['@graph'][index]["rdfs:label"].append({
"@value": translated_str,
"@language": lang
})
changed = True
if not changed:
continue
save_json(ontology_json, filename + '.new')
def _test_can_replyto(base_dir: str) -> None:
print('test_can_reply_to')
system_language = 'en'
languages_understood = [system_language]
nickname = 'test27637'
domain = 'rando.site'
port = 443
http_prefix = 'https'
content = 'This is a test post with links.\n\n' + \
'ftp://ftp.ncdc.noaa.gov/pub/data/ghcn/v4/\n\nhttps://libreserver.org'
save_to_file = False
client_to_server = False
comments_enabled = True
attach_image_filename = None
media_type = None
image_description = None
city = 'London, England'
test_in_reply_to = None
test_in_reply_to_atom_uri = None
test_subject = None
test_schedule_post = False
test_event_date = None
test_event_time = None
test_event_end_time = None
test_location = None
test_is_article = False
conversation_id = None
low_bandwidth = True
content_license_url = 'https://creativecommons.org/licenses/by-nc/4.0'
media_license_url = 'https://creativecommons.org/licenses/by-nc/4.0'
media_creator = 'The Penguin'
translate = {}
buy_url = ''
video_transcript = ''
post_json_object = \
create_public_post(base_dir, nickname, domain, port, http_prefix,
content, save_to_file,
client_to_server, comments_enabled,
attach_image_filename, media_type,
image_description, video_transcript, city,
test_in_reply_to, test_in_reply_to_atom_uri,
test_subject, test_schedule_post,
test_event_date, test_event_time,
test_event_end_time, test_location,
test_is_article, system_language, conversation_id,
low_bandwidth, content_license_url,
media_license_url, media_creator,
languages_understood, translate, buy_url)
# set the date on the post
curr_date_str = "2021-09-08T20:45:00Z"
post_json_object['published'] = curr_date_str
post_json_object['object']['published'] = curr_date_str
# test a post within the reply interval
post_url = post_json_object['object']['id']
reply_interval_hours = 2
curr_date_str = "2021-09-08T21:32:10Z"
assert can_reply_to(base_dir, nickname, domain,
post_url, reply_interval_hours,
curr_date_str,
post_json_object)
# test a post outside of the reply interval
curr_date_str = "2021-09-09T09:24:47Z"
assert not can_reply_to(base_dir, nickname, domain,
post_url, reply_interval_hours,
curr_date_str,
post_json_object)
def _test_seconds_between_publish() -> None:
print('test_seconds_between_published')
published1 = "2021-10-14T09:39:27Z"
published2 = "2021-10-14T09:41:28Z"
seconds_elapsed = seconds_between_published(published1, published2)
assert seconds_elapsed == 121
# invalid date
published2 = "2021-10-14N09:41:28Z"
seconds_elapsed = seconds_between_published(published1, published2)
assert seconds_elapsed == -1
def _test_word_similarity() -> None:
print('test_words_similarity')
min_words = 10
content1 = "This is the same"
content2 = "This is the same"
assert words_similarity(content1, content2, min_words) == 100
content1 = "This is our world now... " + \
"the world of the electron and the switch, the beauty of the baud"
content2 = "This is our world now. " + \
"The world of the electron and the webkit, the beauty of the baud"
similarity = words_similarity(content1, content2, min_words)
assert similarity > 70
content1 = "
We're growing!
A new denizen " + \
"is frequenting HackBucket. You probably know him already " + \
"from her epic typos - but let's not spoil too much " + \
"\ud83d\udd2e
"
content2 = "
We're growing!
A new denizen " + \
"is frequenting HackBucket. You probably know them already " + \
"from their epic typos - but let's not spoil too much " + \
"\ud83d\udd2e
'
if text_bold != expected:
print(text_bold)
assert text_bold == expected
text = "There's the quoted text here"
text_bold = bold_reading_string(text)
expected = \
"There's the quoted text here"
if text_bold != expected:
print(text_bold)
assert text_bold == expected
text = '
'
if text_bold != expected:
print(text_bold)
assert text_bold == expected
def _test_diff_content() -> None:
print('diff_content')
prev_content = \
'Some text before.\n' + \
'Starting sentence. This is some content.\nThis is another line.'
content = \
'Some text before.\nThis is some more content.\nThis is another line.'
result = content_diff(content, prev_content)
expected = \
'
' + \
'
'
assert result == expected
content = \
'Some text before.\nThis is content.\nThis line.'
result = content_diff(content, prev_content)
expected = \
'
'
assert html_str == expected
def _test_missing_theme_colors(base_dir: str) -> None:
print('test_missing_colors')
theme_filename = base_dir + '/theme/default/theme.json'
assert os.path.isfile(theme_filename)
default_theme_json = load_json(theme_filename)
assert default_theme_json
themes = get_themes_list(base_dir)
for theme_name in themes:
if theme_name == 'default':
continue
theme_filename = \
base_dir + '/theme/' + theme_name.lower() + '/theme.json'
if not os.path.isfile(theme_filename):
continue
theme_json = load_json(theme_filename)
if not theme_json:
continue
updated = False
for property, value in default_theme_json.items():
if not theme_json.get(property):
theme_json[property] = value
updated = True
if updated:
save_json(theme_json, theme_filename)
print(theme_name + ' updated')
def _test_color_contrast_value(base_dir: str) -> None:
print('test_color_contrast_value')
minimum_color_contrast = 4.5
background = 'black'
foreground = 'white'
contrast = color_contrast(background, foreground)
assert contrast
assert contrast > 20
assert contrast < 22
foreground = 'grey'
contrast = color_contrast(background, foreground)
assert contrast
assert contrast > 5
assert contrast < 6
themes = get_themes_list(base_dir)
for theme_name in themes:
theme_filename = \
base_dir + '/theme/' + theme_name.lower() + '/theme.json'
if not os.path.isfile(theme_filename):
continue
theme_json = load_json(theme_filename)
if not theme_json:
continue
if not theme_json.get('main-fg-color'):
continue
if not theme_json.get('main-bg-color'):
continue
foreground = theme_json['main-fg-color']
background = theme_json['main-bg-color']
contrast = color_contrast(background, foreground)
if contrast is None:
continue
if contrast < minimum_color_contrast:
print('Theme ' + theme_name + ' has not enough color contrast ' +
str(contrast) + ' < ' + str(minimum_color_contrast))
assert contrast >= minimum_color_contrast
print('Color contrast is ok for all themes')
def _test_remove_end_of_line():
print('remove_end_of_line')
text = 'some text\r\n'
expected = 'some text'
assert remove_eol(text) == expected
text = 'some text'
assert remove_eol(text) == expected
def _test_dogwhistles():
print('dogwhistles')
dogwhistles = {
"X-hamstered": "hamsterism",
"gerbil": "rodent",
"*snake": "slither",
"start*end": "something"
}
content = 'This text does not contain any dogwhistles'
assert not detect_dogwhistles(content, dogwhistles)
content = 'A gerbil named joe'
assert detect_dogwhistles(content, dogwhistles)
content = 'A rattlesnake.'
assert detect_dogwhistles(content, dogwhistles)
content = 'A startthingend.'
assert detect_dogwhistles(content, dogwhistles)
content = 'This content is unhamstered and yhamstered.'
result = detect_dogwhistles(content, dogwhistles)
assert result
assert result.get('hamstered')
assert result['hamstered']['count'] == 2
assert result['hamstered']['category'] == "hamsterism"
def _test_text_standardize():
print('text_standardize')
expected = 'This is a test'
result = standardize_text(expected)
if result != expected:
print(result)
assert result == expected
text = '𝔗𝔥𝔦𝔰 𝔦𝔰 𝔞 𝔱𝔢𝔰𝔱'
result = standardize_text(text)
if result != expected:
print(result)
assert result == expected
text = '𝕿𝖍𝖎𝖘 𝖎𝖘 𝖆 𝖙𝖊𝖘𝖙'
result = standardize_text(text)
if result != expected:
print(result)
assert result == expected
text = '𝓣𝓱𝓲𝓼 𝓲𝓼 𝓪 𝓽𝓮𝓼𝓽'
result = standardize_text(text)
if result != expected:
print(result)
assert result == expected
text = '𝒯𝒽𝒾𝓈 𝒾𝓈 𝒶 𝓉𝑒𝓈𝓉'
result = standardize_text(text)
if result != expected:
print(result)
assert result == expected
text = '𝕋𝕙𝕚𝕤 𝕚𝕤 𝕒 𝕥𝕖𝕤𝕥'
result = standardize_text(text)
if result != expected:
print(result)
assert result == expected
text = 'Ｔｈｉｓ ｉｓ ａ ｔｅｓｔ'
result = standardize_text(text)
if result != expected:
print(result)
assert result == expected
def _test_combine_lines():
print('combine_lines')
text = 'This is a test'
expected = text
result = combine_textarea_lines(text)
if result != expected:
print('expected: ' + expected)
print('result: ' + result)
assert result == expected
text = 'First line.\n\nSecond line.'
expected = 'First line.
Second line.'
result = combine_textarea_lines(text)
if result != expected:
print('expected: ' + expected)
print('result: ' + result)
assert result == expected
text = 'First\nline.\n\nSecond\nline.'
expected = 'First line.
Second line.'
result = combine_textarea_lines(text)
if result != expected:
print('expected: ' + expected)
print('result: ' + result)
assert result == expected
# with extra space
text = 'First\nline.\n\nSecond \nline.'
expected = 'First line.
Second line.'
result = combine_textarea_lines(text)
if result != expected:
print('expected: ' + expected)
print('result: ' + result)
assert result == expected
text = 'Introduction blurb.\n\n* List item 1\n' + \
'* List item 2\n* List item 3\n\nFinal blurb.'
expected = 'Introduction blurb.
* List item 1\n' + \
'* List item 2\n* List item 3
Final blurb.'
result = combine_textarea_lines(text)
if result != expected:
print('expected: ' + expected)
print('result: ' + result)
assert result == expected
def _test_hashtag_maps():
print('hashtag_maps')
content = \
"
This is a test, with a couple of links and a " + \
"#" + \
"Hashtag
' + \
'Compelling description with #ExcitingHashtag, which is ' + \
'being posted in #BoringForum'
tags = get_newswire_tags(rss_description, 10)
assert len(tags) == 2
assert '#BoringForum' in tags
assert '#ExcitingHashtag' in tags
def _test_first_paragraph_from_string():
print('test_first_paragraph_from_string')
test_str = \
'' + \
'
.'
result = markdown_to_html(markdown)
if result != expected:
print(result)
assert result == expected
def _test_extract_text_fields_from_post():
print('test_extract_text_fields_in_post')
boundary = '--LYNX'
form_data = '--LYNX\r\nContent-Disposition: form-data; ' + \
'name="fieldName"\r\nContent-Type: text/plain; ' + \
'charset=utf-8\r\n\r\nThis is a lynx test\r\n' + \
'--LYNX\r\nContent-Disposition: ' + \
'form-data; name="submitYes"\r\nContent-Type: text/plain; ' + \
'charset=utf-8\r\n\r\nBUTTON\r\n--LYNX--\r\n'
debug = True
fields = extract_text_fields_in_post(None, boundary, debug, form_data)
print('fields: ' + str(fields))
assert fields
assert fields['fieldName'] == 'This is a lynx test'
assert fields['submitYes'] == 'BUTTON'
boundary = '-----------------------------116202748023898664511855843036'
form_data = '-----------------------------116202748023898664511855' + \
'843036\r\nContent-Disposition: form-data; name="submitPost"' + \
'\r\n\r\nSubmit\r\n-----------------------------116202748023' + \
'898664511855843036\r\nContent-Disposition: form-data; name=' + \
'"subject"\r\n\r\n\r\n-----------------------------116202748' + \
'023898664511855843036\r\nContent-Disposition: form-data; na' + \
'me="message"\r\n\r\nThis is a ; test\r\n-------------------' + \
'----------116202748023898664511855843036\r\nContent-Disposi' + \
'tion: form-data; name="commentsEnabled"\r\n\r\non\r\n------' + \
'-----------------------116202748023898664511855843036\r\nCo' + \
'ntent-Disposition: form-data; name="eventDate"\r\n\r\n\r\n' + \
'-----------------------------116202748023898664511855843036' + \
'\r\nContent-Disposition: form-data; name="eventTime"\r\n\r' + \
'\n\r\n-----------------------------116202748023898664511855' + \
'843036\r\nContent-Disposition: form-data; name="location"' + \
'\r\n\r\n\r\n-----------------------------116202748023898664' + \
'511855843036\r\nContent-Disposition: form-data; name=' + \
'"imageDescription"\r\n\r\n\r\n-----------------------------' + \
'116202748023898664511855843036\r\nContent-Disposition: ' + \
'form-data; name="attachpic"; filename=""\r\nContent-Type: ' + \
'application/octet-stream\r\n\r\n\r\n----------------------' + \
'-------116202748023898664511855843036--\r\n' + \
'Content-Disposition: form-data; name="importBlocks"; ' + \
'filename="wildebeest_suspend.csv"\r\nContent-Type: ' + \
'text/csv\r\n\r\n#domain,#severity,#reject_media,#reject_reports,' + \
'#public_comment,#obfuscate\nbgp.social,suspend,false,false,' + \
'"Wildebeest",false\ncesko.social,suspend,false,false,' + \
'"Wildebeest",false\ncloudflare.social,suspend,false,false,' + \
'"Wildebeest",false\ndogfood.social,suspend,false,false,' + \
'"Wildebeest",false\ndomo.cafe,suspend,false,false,"Wildebeest",' + \
'false\nemaw.social,suspend,false,false\n\r\n ' + \
'-----------------------------116202748023898664511855843036--\r\n'
debug = False
fields = extract_text_fields_in_post(None, boundary, debug, form_data)
assert fields['submitPost'] == 'Submit'
assert fields['subject'] == ''
assert fields['commentsEnabled'] == 'on'
assert fields['eventDate'] == ''
assert fields['eventTime'] == ''
assert fields['location'] == ''
assert fields['imageDescription'] == ''
assert fields['message'] == 'This is a ; test'
if not fields['importBlocks'][1:].startswith('#domain,#severity,'):
print(fields['importBlocks'])
assert fields['importBlocks'][1:].startswith('#domain,#severity,')
def _test_speaker_replace_link():
http_prefix = 'https'
nickname = 'mynick'
domain = 'mydomain'
domain_full = domain
print('testSpeakerReplaceLinks')
text = 'The Tor Project: For Snowflake volunteers: If you use ' + \
'Firefox, Brave, or Chrome, our Snowflake extension turns ' + \
'your browser into a proxy that connects Tor users in ' + \
'censored regions to the Tor network. Note: you should ' + \
'not run more than one snowflake in the same ' + \
'network.https://support.torproject.org/censorship/' + \
'how-to-help-running-snowflake/'
detected_links = []
result = \
speaker_replace_links(http_prefix, nickname, domain, domain_full,
text, {'Linked': 'Web link'}, detected_links)
assert len(detected_links) == 1
assert detected_links[0] == \
'https://support.torproject.org/censorship/' + \
'how-to-help-running-snowflake/'
assert 'Web link support.torproject.org' in result
remote_link = 'https://somedomain/tags/sometag'
text = 'Test with a hashtag ' + remote_link + ' link'
detected_links = []
result = \
speaker_replace_links(http_prefix, nickname, domain, domain_full,
text, {'Linked': 'Web link'}, detected_links)
assert len(detected_links) == 1
local_link = \
'https://' + domain_full + '/users/' + nickname + \
'?remotetag=' + remote_link.replace('/', '--')
assert detected_links[0] == local_link
assert 'Web link somedomain' in result
def _test_camel_case_split():
print('test_camel_case_split')
test_str = 'ThisIsCamelCase'
assert camel_case_split(test_str) == 'This Is Camel Case'
test_str = 'Notcamelcase test'
assert camel_case_split(test_str) == 'Notcamelcase test'
def _test_emoji_images():
print('test_emoji_images')
emoji_filename = 'emoji/default_emoji.json'
assert os.path.isfile(emoji_filename)
emoji_json = load_json(emoji_filename)
assert emoji_json
for emoji_name, emoji_image in emoji_json.items():
emoji_image_filename = 'emoji/' + emoji_image + '.png'
if not os.path.isfile(emoji_image_filename):
print('Missing emoji image ' + emoji_name + ' ' +
emoji_image + '.png')
assert os.path.isfile(emoji_image_filename)
def _test_extract_pgp_public_key():
print('testExtractPGPPublicKey')
pub_key = \
'-----BEGIN PGP PUBLIC KEY BLOCK-----\n\n' + \
'mDMEWZBueBYJKwYBBAHaRw8BAQdAKx1t6wL0RTuU6/' + \
'IBjngMbVJJ3Wg/3UW73/PV\n' + \
'I47xKTS0IUJvYiBNb3R0cmFtIDxib2JAZnJlZWRvb' + \
'WJvbmUubmV0PoiQBBMWCAA4\n' + \
'FiEEmruCwAq/OfgmgEh9zCU2GR+nwz8FAlmQbngCG' + \
'wMFCwkIBwMFFQoJCAsFFgID\n' + \
'AQACHgECF4AACgkQzCU2GR+nwz/9sAD/YgsHnVszH' + \
'Nz1zlVc5EgY1ByDupiJpHj0\n' + \
'XsLYk3AbNRgBALn45RqgD4eWHpmOriH09H5Rc5V9i' + \
'N4+OiGUn2AzJ6oHuDgEWZBu\n' + \
'eBIKKwYBBAGXVQEFAQEHQPRBG2ZQJce475S3e0Dxe' + \
'b0Fz5WdEu2q3GYLo4QG+4Ry\n' + \
'AwEIB4h4BBgWCAAgFiEEmruCwAq/OfgmgEh9zCU2G' + \
'R+nwz8FAlmQbngCGwwACgkQ\n' + \
'zCU2GR+nwz+OswD+JOoyBku9FzuWoVoOevU2HH+bP' + \
'OMDgY2OLnST9ZSyHkMBAMcK\n' + \
'fnaZ2Wi050483Sj2RmQRpb99Dod7rVZTDtCqXk0J\n' + \
'=gv5G\n' + \
'-----END PGP PUBLIC KEY BLOCK-----'
test_str = "Some introduction\n\n" + pub_key + "\n\nSome message."
assert contains_pgp_public_key(test_str)
assert not contains_pgp_public_key('String without a pgp key')
result = extract_pgp_public_key(test_str)
assert result
assert result == pub_key
def test_update_actor(base_dir: str):
print('Testing update of actor properties')
global TEST_SERVER_ALICE_RUNNING
TEST_SERVER_ALICE_RUNNING = False
http_prefix = 'http'
proxy_type = None
federation_list = []
if os.path.isdir(base_dir + '/.tests'):
shutil.rmtree(base_dir + '/.tests',
ignore_errors=False, onerror=None)
os.mkdir(base_dir + '/.tests')
# create the server
alice_dir = base_dir + '/.tests/alice'
alice_domain = '127.0.0.11'
alice_port = 61792
alice_send_threads = []
bob_address = '127.0.0.84:6384'
global THR_ALICE
if THR_ALICE:
while THR_ALICE.is_alive():
THR_ALICE.stop()
time.sleep(1)
THR_ALICE.kill()
THR_ALICE = \
thread_with_trace(target=create_server_alice,
args=(alice_dir, alice_domain,
alice_port, bob_address,
federation_list, False, False,
alice_send_threads),
daemon=True)
THR_ALICE.start()
assert THR_ALICE.is_alive() is True
# wait for server to be running
ctr = 0
while not TEST_SERVER_ALICE_RUNNING:
time.sleep(1)
ctr += 1
if ctr > 60:
break
print('Alice online: ' + str(TEST_SERVER_ALICE_RUNNING))
print('\n\n*******************************************************')
print('Alice updates her PGP key')
session_alice = create_session(proxy_type)
cached_webfingers = {}
person_cache = {}
password = 'alicepass'
outbox_path = alice_dir + '/accounts/alice@' + alice_domain + '/outbox'
actor_filename = \
alice_dir + '/accounts/' + 'alice@' + alice_domain + '.json'
assert os.path.isfile(actor_filename)
assert len([name for name in os.listdir(outbox_path)
if os.path.isfile(os.path.join(outbox_path, name))]) == 0
pub_key = \
'-----BEGIN PGP PUBLIC KEY BLOCK-----\n\n' + \
'mDMEWZBueBYJKwYBBAHaRw8BAQdAKx1t6wL0RTuU6/' + \
'IBjngMbVJJ3Wg/3UW73/PV\n' + \
'I47xKTS0IUJvYiBNb3R0cmFtIDxib2JAZnJlZWRvb' + \
'WJvbmUubmV0PoiQBBMWCAA4\n' + \
'FiEEmruCwAq/OfgmgEh9zCU2GR+nwz8FAlmQbngCG' + \
'wMFCwkIBwMFFQoJCAsFFgID\n' + \
'AQACHgECF4AACgkQzCU2GR+nwz/9sAD/YgsHnVszH' + \
'Nz1zlVc5EgY1ByDupiJpHj0\n' + \
'XsLYk3AbNRgBALn45RqgD4eWHpmOriH09H5Rc5V9i' + \
'N4+OiGUn2AzJ6oHuDgEWZBu\n' + \
'eBIKKwYBBAGXVQEFAQEHQPRBG2ZQJce475S3e0Dxe' + \
'b0Fz5WdEu2q3GYLo4QG+4Ry\n' + \
'AwEIB4h4BBgWCAAgFiEEmruCwAq/OfgmgEh9zCU2G' + \
'R+nwz8FAlmQbngCGwwACgkQ\n' + \
'zCU2GR+nwz+OswD+JOoyBku9FzuWoVoOevU2HH+bP' + \
'OMDgY2OLnST9ZSyHkMBAMcK\n' + \
'fnaZ2Wi050483Sj2RmQRpb99Dod7rVZTDtCqXk0J\n' + \
'=gv5G\n' + \
'-----END PGP PUBLIC KEY BLOCK-----'
signing_priv_key_pem = None
actor_update = \
pgp_public_key_upload(alice_dir, session_alice,
'alice', password,
alice_domain, alice_port,
http_prefix,
cached_webfingers, person_cache,
True, pub_key, signing_priv_key_pem)
print('actor update result: ' + str(actor_update))
assert actor_update
# load alice actor
print('Loading actor: ' + actor_filename)
actor_json = load_json(actor_filename)
assert actor_json
if len(actor_json['attachment']) == 0:
print("actor_json['attachment'] has no contents")
assert len(actor_json['attachment']) > 0
property_found = False
for property_value in actor_json['attachment']:
if property_value['name'] == 'PGP':
print('PGP property set within attachment')
assert pub_key in property_value['value']
property_found = True
assert property_found
# stop the server
THR_ALICE.kill()
THR_ALICE.join()
assert THR_ALICE.is_alive() is False
os.chdir(base_dir)
if os.path.isdir(base_dir + '/.tests'):
shutil.rmtree(base_dir + '/.tests', ignore_errors=False, onerror=None)
def _test_remove_interactions() -> None:
print('test_remove_post_interactions')
post_json_object = {
"type": "Create",
"object": {
"to": ["#Public"],
"likes": {
"items": ["a", "b", "c"]
},
"replies": {
"replyStuff": ["a", "b", "c"]
},
"shares": {
"sharesStuff": ["a", "b", "c"]
},
"bookmarks": {
"bookmarksStuff": ["a", "b", "c"]
},
"ignores": {
"ignoresStuff": ["a", "b", "c"]
}
}
}
remove_post_interactions(post_json_object, True)
assert post_json_object['object']['likes']['items'] == []
assert post_json_object['object']['replies'] == {}
assert post_json_object['object']['shares'] == {}
assert post_json_object['object']['bookmarks'] == {}
assert post_json_object['object']['ignores'] == {}
post_json_object['object']['to'] = ["some private address"]
assert not remove_post_interactions(post_json_object, False)
def _test_spoofed_geolocation() -> None:
print('test_spoof_geolocation')
nogo_line = \
'NEW YORK, USA: 73.951W,40.879, 73.974W,40.83, ' + \
'74.029W,40.756, 74.038W,40.713, 74.056W,40.713, ' + \
'74.127W,40.647, 74.038W,40.629, 73.995W,40.667, ' + \
'74.014W,40.676, 73.994W,40.702, 73.967W,40.699, ' + \
'73.958W,40.729, 73.956W,40.745, 73.918W,40.781, ' + \
'73.937W,40.793, 73.946W,40.782, 73.977W,40.738, ' + \
'73.98W,40.713, 74.012W,40.705, 74.006W,40.752, ' + \
'73.955W,40.824'
polygon = parse_nogo_string(nogo_line)
assert len(polygon) > 0
assert polygon[0][1] == -73.951
assert polygon[0][0] == 40.879
cities_list = [
'NEW YORK, USA:40.7127281:W74.0060152:784',
'LOS ANGELES, USA:34.0536909:W118.242766:1214',
'SAN FRANCISCO, USA:37.74594738515095:W122.44299445520019:121',
'HOUSTON, USA:29.6072:W95.1586:1553',
'MANCHESTER, ENGLAND:53.4794892:W2.2451148:1276',
'BERLIN, GERMANY:52.5170365:13.3888599:891',
'ANKARA, TURKEY:39.93:32.85:24521',
'LONDON, ENGLAND:51.5073219:W0.1276474:1738',
'SEATTLE, USA:47.59840153253106:W122.31143714060059:217'
]
test_square = [
[[0.03, 0.01], [0.02, 10], [10.01, 10.02], [10.03, 0.02]]
]
assert point_in_nogo(test_square, 5, 5)
assert point_in_nogo(test_square, 2, 3)
assert not point_in_nogo(test_square, 20, 5)
assert not point_in_nogo(test_square, 11, 6)
assert not point_in_nogo(test_square, 5, -5)
assert not point_in_nogo(test_square, 5, 11)
assert not point_in_nogo(test_square, -5, -5)
assert not point_in_nogo(test_square, -5, 5)
nogo_list = []
curr_time = datetime.datetime.utcnow()
decoy_seed = 7634681
city_radius = 0.1
coords = spoof_geolocation('', 'los angeles', curr_time,
decoy_seed, cities_list, nogo_list)
assert coords[0] >= 34.0536909 - city_radius
assert coords[0] <= 34.0536909 + city_radius
assert coords[1] >= 118.242766 - city_radius
assert coords[1] <= 118.242766 + city_radius
assert coords[2] == 'N'
assert coords[3] == 'W'
assert len(coords[4]) > 4
assert len(coords[5]) > 4
assert coords[6] > 0
nogo_list = []
coords = spoof_geolocation('', 'unknown', curr_time,
decoy_seed, cities_list, nogo_list)
assert coords[0] >= 51.8744 - city_radius
assert coords[0] <= 51.8744 + city_radius
assert coords[1] >= 0.368333 - city_radius
assert coords[1] <= 0.368333 + city_radius
assert coords[2] == 'N'
assert coords[3] == 'W'
assert len(coords[4]) == 0
assert len(coords[5]) == 0
assert coords[6] == 0
kml_str = '\n'
kml_str += '