This is a valid message
' 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... ' + \ '
' assert dangerous_markup(content, allow_local_network_access) content = 'This is a valid-looking message. But wait... ' + \ '<script>document.getElementById("concentrated")' + \ '.innerHTML = "evil";</script>
' assert dangerous_markup(content, allow_local_network_access) content = 'This html contains more than you expected... ' + \ '
' assert dangerous_markup(content, allow_local_network_access) content = 'This is a valid-looking message. But wait... ' + \ '
' assert dangerous_markup(content, allow_local_network_access) content = 'This is a valid-looking message. But it contains ' + \
'spyware.
This is a valid-looking message. But it contains ' + \ 'spyware.
' assert dangerous_markup(content, allow_local_network_access) content = 'This message embeds an evil frame.' + \ '
' assert dangerous_markup(content, allow_local_network_access) content = 'This message tries to obfuscate an evil frame.' + \ '< iframe src = "somesite">
' assert dangerous_markup(content, allow_local_network_access) content = 'This message is not necessarily evil, but annoying.' + \ '
' assert dangerous_markup(content, allow_local_network_access) content = '
This message contans a ' + \ 'valid link.
' assert not dangerous_markup(content, allow_local_network_access) content = 'This message contans a ' + \ '' + \ 'valid link having invalid but harmless name.
' assert not dangerous_markup(content, allow_local_network_access) content = 'This message which ' + \ 'tries to access the local network
' assert dangerous_markup(content, allow_local_network_access) content = 'This message which ' + \ 'tries to access the local network
' assert dangerous_markup(content, allow_local_network_access) content = '127.0.0.1 This message which does not access ' + \ 'the local network
' assert not dangerous_markup(content, allow_local_network_access) def _run_html_replace_quote_marks(): print('html_replace_quote_marks') test_str = 'The "cat" "sat" on the mat' result = html_replace_quote_marks(test_str) assert result == 'The “cat” “sat” on the mat' test_str = 'The cat sat on the mat' result = html_replace_quote_marks(test_str) assert result == 'The cat sat on the mat' test_str = '"hello"' result = html_replace_quote_marks(test_str) assert result == '“hello”' test_str = '"hello" "test" html' result = html_replace_quote_marks(test_str) assert result == '“hello” “test” html' def _test_json_post_allows_comment(): print('test_json_post_allows_comments') post_json_object = { "id": "123" } assert json_post_allows_comments(post_json_object) post_json_object = { "id": "123", "commentsEnabled": False } assert not json_post_allows_comments(post_json_object) post_json_object = { "id": "123", "rejectReplies": False } assert json_post_allows_comments(post_json_object) post_json_object = { "id": "123", "rejectReplies": True } assert not json_post_allows_comments(post_json_object) post_json_object = { "id": "123", "commentsEnabled": True } assert json_post_allows_comments(post_json_object) post_json_object = { "id": "123", "object": { "commentsEnabled": True } } assert json_post_allows_comments(post_json_object) post_json_object = { "id": "123", "object": { "commentsEnabled": False } } assert not json_post_allows_comments(post_json_object) def _test_remove_id_ending(): print('test_remove_id_ending') test_str = 'https://activitypub.somedomain.net' result_str = remove_id_ending(test_str) assert result_str == 'https://activitypub.somedomain.net' test_str = \ 'https://activitypub.somedomain.net/users/foo/' + \ 'statuses/34544814814/activity' result_str = remove_id_ending(test_str) assert result_str == \ 'https://activitypub.somedomain.net/users/foo/statuses/34544814814' test_str = \ 'https://undo.somedomain.net/users/foo/statuses/34544814814/undo' result_str = remove_id_ending(test_str) assert result_str == \ 'https://undo.somedomain.net/users/foo/statuses/34544814814' test_str = \ 'https://event.somedomain.net/users/foo/statuses/34544814814/event' result_str = remove_id_ending(test_str) assert result_str == \ 'https://event.somedomain.net/users/foo/statuses/34544814814' def _test_valid_content_warning(): print('test_valid_content_warning') result_str = valid_content_warning('Valid content warning') assert result_str == 'Valid content warning' result_str = valid_content_warning('Invalid #content warning') assert result_str == 'Invalid content warning' result_str = \ valid_content_warning('Invalid content warning') assert result_str == 'Invalid content warning' def _test_translation_labels() -> None: print('test_translation_labels') lang_json = load_json('translations/en.json') assert lang_json 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 'translate[' not in source_str: continue sections = source_str.split('translate[') ctr = 0 for text in sections: if ctr == 0: ctr += 1 continue if ']' not in text: continue label_str = text.split(']')[0] if '"' not in label_str and "'" not in label_str: continue if '+' in label_str: continue label_str = label_str[1:] label_str = label_str[:len(label_str)-1] assert label_str if not lang_json.get(label_str): print("Translation for label '" + label_str + "' " + "in module " + source_file + " was not found") assert False break def _test_translations(base_dir: str) -> None: print('test_translations') languages_str = get_supported_languages(base_dir) assert languages_str # load all translations into a dict lang_dict = {} for lang in languages_str: lang_json = load_json('translations/' + lang + '.json') if not lang_json: print('Missing language file ' + 'translations/' + lang + '.json') assert lang_json lang_dict[lang] = lang_json # load english translations translations_json = load_json('translations/en.json') # test each english string exists in the other language files for english_str, _ in translations_json.items(): for lang in languages_str: lang_json = lang_dict[lang] if not lang_json.get(english_str): print(english_str + ' is missing from ' + lang + '.json') assert lang_json.get(english_str) def _test_constant_time_string(): print('test_constant_time_string_check') assert constant_time_string_check('testing', 'testing') assert not constant_time_string_check('testing', '1234') assert not constant_time_string_check('testing', '1234567') itterations = 256 start = time.time() test_str = 'nnjfbefefbsnjsdnvbcueftqfeuqfbqefnjeniwufgy' for _ in range(itterations): constant_time_string_check(test_str, test_str) end = time.time() av_time1 = ((end - start) * 1000000 / itterations) # change a single character and observe timing difference start = time.time() for _ in range(itterations): constant_time_string_check(test_str, test_str) end = time.time() av_time2 = ((end - start) * 1000000 / itterations) time_diff_microseconds = abs(av_time2 - av_time1) # time difference should be less than 10uS assert int(time_diff_microseconds) < 10 # change multiple characters and observe timing difference start = time.time() test_str2 = 'ano1befffbsn7sd3vbluef6qseuqfpqeznjgni9bfgi' for _ in range(itterations): constant_time_string_check(test_str, test_str2) end = time.time() av_time2 = ((end - start) * 1000000 / itterations) time_diff_microseconds = abs(av_time2 - av_time1) # time difference should be less than 10uS assert int(time_diff_microseconds) < 10 def _test_replace_email_quote(): print('test_replace_email_quote') test_str = 'This content has no quote.
' assert html_replace_email_quote(test_str) == test_str test_str = 'This content has no quote.
' + \ 'With multiple
lines
' assert html_replace_email_quote(test_str) == test_str test_str = '"This is a quoted paragraph."
' assert html_replace_email_quote(test_str) == \ 'This is a quoted paragraph.' test_str = "
" + \
"@nickname " + \
"
> This is a quote
Some other text.
" expected_str = "" + \
"@nickname " + \
"
This is a quote" + \ "
Some other text.
" result_str = html_replace_email_quote(test_str) if result_str != expected_str: print('Result: ' + str(result_str)) print('Expect: ' + expected_str) assert result_str == expected_str test_str = "Some text:
> first line->second line
" + \ "Some question?
" expected_str = "Some text:
first line-
" + \ "second line
Some question?
" result_str = html_replace_email_quote(test_str) if result_str != expected_str: print('Result: ' + str(result_str)) print('Expect: ' + expected_str) assert result_str == expected_str test_str = "" + \ "@somenick" + \ "
> Text1.
>
" + \
"> Text2
>
> Text3
" + \
">
> Text4
>
> " + \
"Text5
>
> Text6
Text7
" expected_str = "" + \ "@somenick" + \ "
Text1.
" + \ "Text2
Text3
>
Text4
" + \ "
Text5
Text6
Text7
" result_str = html_replace_email_quote(test_str) if result_str != expected_str: print('Result: ' + str(result_str)) print('Expect: ' + expected_str) assert result_str == expected_str def _test_strip_html_tag(): print('test_remove_html_tag') test_str = "
' + \
'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 = \
'' + \
'This is another paragraph
' result_str = first_paragraph_from_string(test_str) if result_str != 'This is a test': print(result_str) assert result_str == 'This is a test' test_str = 'Testing without html' result_str = first_paragraph_from_string(test_str) assert result_str == test_str def _test_parse_newswire_feed_date(): print('test_parse_feed_date') unique_string_identifier = 'some string abcd' pub_date = "2020-12-14T00:08:06+00:00" published_date = parse_feed_date(pub_date, unique_string_identifier) assert published_date == "2020-12-14 00:08:06+00:00" pub_date = "Tue, 08 Dec 2020 06:24:38 -0600" published_date = parse_feed_date(pub_date, unique_string_identifier) assert published_date == "2020-12-08 12:24:38+00:00" pub_date = "2020-08-27T16:12:34+00:00" published_date = parse_feed_date(pub_date, unique_string_identifier) assert published_date == "2020-08-27 16:12:34+00:00" pub_date = "Sun, 22 Nov 2020 19:51:33 +0100" published_date = parse_feed_date(pub_date, unique_string_identifier) assert published_date == "2020-11-22 18:51:33+00:00" pub_date = "Sun, 22 Nov 2020 00:00:00 +0000" published_date = parse_feed_date(pub_date, unique_string_identifier) assert published_date != "2020-11-22 00:00:00+00:00" assert "2020-11-22 00:" in published_date def _test_valid_nick(): print('test_valid_nickname') domain = 'somedomain.net' nickname = 'myvalidnick' assert valid_nickname(domain, nickname) nickname = 'my.invalid.nick' assert not valid_nickname(domain, nickname) nickname = 'myinvalidnick?' assert not valid_nickname(domain, nickname) nickname = 'my invalid nick?' assert not valid_nickname(domain, nickname) def _test_guess_tag_category() -> None: print('test_guess_hashtag_category') hashtag_categories = { "foo": ["swan", "goose"], "bar": ["cats", "mouse"] } guess = guess_hashtag_category("unspecifiedgoose", hashtag_categories) assert guess == "foo" guess = guess_hashtag_category("mastocats", hashtag_categories) assert guess == "bar" def _test_mentioned_people(base_dir: str) -> None: print('test_get_mentioned_people') content = "@dragon@cave.site @bat@cave.site This is a test." actors = \ get_mentioned_people(base_dir, 'https', content, 'mydomain', False) assert actors assert len(actors) == 2 assert actors[0] == "https://cave.site/users/dragon" assert actors[1] == "https://cave.site/users/bat" def _test_reply_to_public_post(base_dir: str) -> None: system_language = 'en' languages_understood = [system_language] nickname = 'test7492362' domain = 'other.site' port = 443 http_prefix = 'https' post_id = \ http_prefix + '://rat.site/users/ninjarodent/statuses/63746173435' content = "@ninjarodent@rat.site This is a test." save_to_file = False client_to_server = False comments_enabled = True attach_image_filename = None media_type = None image_description = 'Some description' city = 'London, England' test_in_reply_to = post_id 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/4.0' translate = {} reply = \ 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, 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, languages_understood, translate) # print(str(reply)) expected_str = \ '' + \ '@ninjarodent' + \ ' This is a test.
' 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/4.0' translate = {} 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, 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, languages_understood, translate) expected_str = \ 'This is a test post with links.
' + \
'' + \
'ftp://' + \
'' + \
'ftp.ncdc.noaa.gov/pub/data/ghcn/v4/' + \
'
' + \
'https://' + \
'libreserver.org
Some text
Other text
More text
" + \ "Errno::EOHNOES (No such file or rodent @ " + \
"ik_right - /tmp/blah.png)
" + \ "(" + \ "wuh)
Oh yeah like for sure
" + \ "Ground sloth tin opener
" + \ "" post_json_object = \ create_public_post(base_dir, nickname, domain, port, http_prefix, content, False, False, True, None, None, False, None, 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, languages_understood, translate) assert post_json_object['object']['content'] == content assert post_json_object['object']['contentMap'][system_language] == content def _test_mastoapi(): print('test_masto_api') nickname = 'ThisIsATestNickname' masto_id = get_masto_api_v1id_from_nickname(nickname) assert masto_id nickname2 = get_nickname_from_masto_api_v1id(masto_id) if nickname2 != nickname: print(nickname + ' != ' + nickname2) assert nickname2 == nickname def _test_domain_handling(): print('test_domain_handling') test_domain = 'localhost' assert decoded_host(test_domain) == test_domain test_domain = '127.0.0.1:60' assert decoded_host(test_domain) == test_domain test_domain = '192.168.5.153' assert decoded_host(test_domain) == test_domain test_domain = 'xn--espaa-rta.icom.museum' assert decoded_host(test_domain) == "españa.icom.museum" def _test_prepare_html_post_nick(): print('test_prepare_html_post_nickname') post_html = ' Some quote or other' expected = \ 'This is a quotation:\n' + \ '
Some quote or other' result = markdown_to_html(markdown) if result != expected: print(result) assert result == expected markdown = 'This is a multi-line quotation:\n' + \ '> The first line\n' + \ '> The second line' assert markdown_to_html(markdown) == \ 'This is a multi-line quotation:
\n' + \ '
The first line The second line' markdown = 'This is a list of points:\n' + \ ' * Point 1\n' + \ ' * Point 2\n\n' + \ 'And some other text.' result = markdown_to_html(markdown) expected = \ 'This is a list of points:
\n
- ' + \
'\n
- Point 1 \n' + \ '
- Point 2 \n\n
\n' + \ 'And some other text.
\n' if result != expected: print(result) assert result == expected markdown = 'This is a list of points:\n' + \ ' * **Point 1**\n' + \ ' * *Point 2*\n\n' + \ 'And some other text.' result = markdown_to_html(markdown) expected = \ 'This is a list of points:
\n
- \n' + \
'
- Point 1 \n' + \ '
- Point 2 \n\n
\n' + \ 'And some other text.
\n' if result != expected: print(result) assert result == expected markdown = 'This is a code section:\n' + \ '``` json\n' + \ '10 PRINT "YOLO"\n' + \ '20 GOTO 10\n' + \ '```\n\n' + \ 'And some other text.' result = markdown_to_html(markdown) expected = \ 'This is a code section:
\n' + \ '
\n' + \
'10 PRINT "YOLO"\n' + \
'20 GOTO 10\n' + \
'\n' + \
'\n' + \ 'And some other text.
\n' if result != expected: print(result) assert result == expected markdown = 'This is **bold**' assert markdown_to_html(markdown) == 'This is bold' markdown = 'This is *italic*' assert markdown_to_html(markdown) == 'This is italic' markdown = 'This is _underlined_' assert markdown_to_html(markdown) == 'This is underlined' markdown = 'This is **just** plain text' assert markdown_to_html(markdown) == 'This is just plain text' markdown = '# Title1\n### Title3\n## Title2\n' expected = '
Title1
\nTitle3
\nTitle2
\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'
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'
def _test_speaker_replace_link():
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(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
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 += '' + 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/4.0' translate = {} 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, 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, languages_understood, translate) # 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
" similarity = words_similarity(content1, content2, min_words) assert similarity > 80 def _test_add_cw_lists(base_dir: str) -> None: print('test_add_CW_from_lists') translate = {} system_language = "en" cw_lists = load_cw_lists(base_dir, True) assert cw_lists post_json_object = { "object": { "sensitive": False, "summary": None, "content": "" } } add_cw_from_lists(post_json_object, cw_lists, translate, 'Murdoch press', system_language) assert post_json_object['object']['sensitive'] is False assert post_json_object['object']['summary'] is None post_json_object = { "object": { "sensitive": False, "summary": None, "contentMap": { "en": "Blah blah news.co.uk blah blah" } } } add_cw_from_lists(post_json_object, cw_lists, translate, 'Murdoch press', system_language) assert post_json_object['object']['sensitive'] is True assert post_json_object['object']['summary'] == "Murdoch Press" post_json_object = { "object": { "sensitive": True, "summary": "Existing CW", "content": "Blah blah news.co.uk blah blah" } } add_cw_from_lists(post_json_object, cw_lists, translate, 'Murdoch press', system_language) assert post_json_object['object']['sensitive'] is True assert post_json_object['object']['summary'] == \ "Murdoch Press / Existing CW" def _test_valid_emoji_content() -> None: print('test_valid_emoji_content') assert not valid_emoji_content(None) assert not valid_emoji_content(' ') assert not valid_emoji_content('j') assert not valid_emoji_content('😀😀😀') assert valid_emoji_content('😀') assert valid_emoji_content('😄') def _test_httpsig_base_new(with_digest: bool, base_dir: str, algorithm: str, digest_algorithm: str) -> None: print('test_httpsig_new(' + str(with_digest) + ')') debug = True path = base_dir + '/.testHttpsigBaseNew' if os.path.isdir(path): shutil.rmtree(path, ignore_errors=False, onerror=None) os.mkdir(path) os.chdir(path) content_type = 'application/activity+json' nickname = 'socrates' host_domain = 'someother.instance' domain = 'argumentative.social' http_prefix = 'https' port = 5576 password = 'SuperSecretPassword' private_key_pem, public_key_pem, _, _ = \ create_person(path, nickname, domain, port, http_prefix, False, False, password) assert private_key_pem if with_digest: message_body_json = { "a key": "a value", "another key": "A string", "yet another key": "Another string" } message_body_json_str = json.dumps(message_body_json) else: message_body_json_str = '' headers_domain = get_full_domain(host_domain, port) date_str = strftime("%a, %d %b %Y %H:%M:%S %Z", gmtime()) boxpath = '/inbox' if not with_digest: headers = { 'host': headers_domain, 'date': date_str, 'accept': content_type } signature_index_header, signature_header = \ sign_post_headers_new(date_str, private_key_pem, nickname, domain, port, host_domain, port, boxpath, http_prefix, message_body_json_str, algorithm, digest_algorithm, debug) else: digest_prefix = get_digest_prefix(digest_algorithm) body_digest = \ message_content_digest(message_body_json_str, digest_algorithm) content_length = len(message_body_json_str) headers = { 'host': headers_domain, 'date': date_str, 'digest': f'{digest_prefix}={body_digest}', 'content-type': content_type, 'content-length': str(content_length) } assert get_digest_algorithm_from_headers(headers) == digest_algorithm signature_index_header, signature_header = \ sign_post_headers_new(date_str, private_key_pem, nickname, domain, port, host_domain, port, boxpath, http_prefix, message_body_json_str, algorithm, digest_algorithm, debug) headers['signature'] = signature_header headers['signature-input'] = signature_index_header print('headers: ' + str(headers)) getreq_method = not with_digest debug = True assert verify_post_headers(http_prefix, public_key_pem, headers, boxpath, getreq_method, None, message_body_json_str, debug) debug = False if with_digest: # everything correct except for content-length headers['content-length'] = str(content_length + 2) assert verify_post_headers(http_prefix, public_key_pem, headers, boxpath, getreq_method, None, message_body_json_str, debug) is False assert verify_post_headers(http_prefix, public_key_pem, headers, '/parambulator' + boxpath, getreq_method, None, message_body_json_str, debug) is False assert verify_post_headers(http_prefix, public_key_pem, headers, boxpath, not getreq_method, None, message_body_json_str, debug) is False if not with_digest: # fake domain headers = { 'host': 'bogon.domain', 'date': date_str, 'content-type': content_type } else: # correct domain but fake message message_body_json_str = \ '{"a key": "a value", "another key": "Fake GNUs", ' + \ '"yet another key": "More Fake GNUs"}' content_length = len(message_body_json_str) digest_prefix = get_digest_prefix(digest_algorithm) body_digest = \ message_content_digest(message_body_json_str, digest_algorithm) headers = { 'host': domain, 'date': date_str, 'digest': f'{digest_prefix}={body_digest}', 'content-type': content_type, 'content-length': str(content_length) } assert get_digest_algorithm_from_headers(headers) == digest_algorithm headers['signature'] = signature_header headers['signature-input'] = signature_index_header pprint(headers) assert verify_post_headers(http_prefix, public_key_pem, headers, boxpath, not getreq_method, None, message_body_json_str, False) is False os.chdir(base_dir) shutil.rmtree(path, ignore_errors=False, onerror=None) def _test_get_actor_from_in_reply_to() -> None: print('test_get_actor_from_in_reply_to') in_reply_to = \ 'https://fosstodon.org/users/bashrc/statuses/107400700612621140' reply_actor = get_actor_from_in_reply_to(in_reply_to) assert reply_actor == 'https://fosstodon.org/users/bashrc' in_reply_to = 'https://fosstodon.org/activity/107400700612621140' reply_actor = get_actor_from_in_reply_to(in_reply_to) assert reply_actor is None def _test_xml_podcast_dict(base_dir: str) -> None: print('test_xml_podcast_dict') xml_str = \ '\n' + \ 'Some text including a link https://some.site/some-path
' expected_text = 'Some text including a link ' + \ '' not in safe_text assert '' not in safe_text web_text = 'Some text with ' expected_text = 'Some text with some script' safe_text = safe_web_text(web_text) if expected_text != safe_text: print('Original html: ' + web_text) print('Expected html: ' + expected_text) print('Actual html: ' + safe_text) assert expected_text == safe_text def _test_published_to_local_timezone() -> None: print('published_to_local_timezone') published_str = '2022-02-25T20:15:00Z' timezone = 'Europe/Berlin' published = \ datetime.datetime.strptime(published_str, "%Y-%m-%dT%H:%M:%SZ") datetime_object = \ convert_published_to_local_timezone(published, timezone) local_time_str = datetime_object.strftime("%a %b %d, %H:%M") assert local_time_str == 'Fri Feb 25, 21:15' timezone = 'Asia/Seoul' published = \ datetime.datetime.strptime(published_str, "%Y-%m-%dT%H:%M:%SZ") datetime_object = \ convert_published_to_local_timezone(published, timezone) local_time_str = datetime_object.strftime("%a %b %d, %H:%M") assert local_time_str == 'Sat Feb 26, 05:15' def _test_bold_reading() -> None: print('bold_reading') text = "This is a test of emboldening." text_bold = bold_reading_string(text) expected = \ "This is a test of " + \ "emboldening." if text_bold != expected: print(text_bold) assert text_bold == expected text = "This is a test of emboldening with paragraph.
" text_bold = bold_reading_string(text) expected = \ "
This is a test of " + \ "emboldening with paragraph.
" if text_bold != expected: print(text_bold) assert text_bold == expected text = \ "This is a test of emboldening
" + \ "With more than one paragraph.
" text_bold = bold_reading_string(text) expected = \ "
This is a test of " + \ "emboldening
With more " + \ "than one paragraph.
" if text_bold != expected: print(text_bold) assert text_bold == expected text = '' text_bold = bold_reading_string(text) expected = \ '
' 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 = '@Someone or other' + \ ' some text
' text_bold = bold_reading_string(text) expected = \ '' + \ '@Someone or other' + \ ' some 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 = \ '
' + \
'
' + \
'
' + \
'
' + \
'
' + \
'
This is some content.
" + \ "Some other content.
" content2 = \ "This is some previous content.
" + \ "Some other previous content.
" content3 = \ "This is some more previous content.
" + \ "Some other previous content.
" post_json_object = { "object": { "content": content1, "published": "2020-12-14T00:08:06Z" } } edits_json = { "2020-12-14T00:05:19Z": { "object": { "content": content3, "published": "2020-12-14T00:05:19Z" } }, "2020-12-14T00:07:34Z": { "object": { "contentMap": { "en": content2 }, "published": "2020-12-14T00:07:34Z" } } } html_str = \ create_edits_html(edits_json, post_json_object, translate, timezone, system_language) assert html_str expected = \ 'SHOW EDITS
' + \ 'Mon Dec 14, 01:07
' + \
'
' + \
'
Mon Dec 14, 01:05
' + \
'
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
" + \
"https://" + \
"" + \
"www.openstreetmap.org/#map=19/52.90860/-3.59917
" + \
"" + \
"https://" + \
"www.google.com/maps/@52.217291,-3.081186" + \
"5,20.04z
" + \
"#" + \
"AnotherHashtag
Some ordinary text
ʇsǝʇ ɐ sı sıɥʇ
' expected = "Some ordinary text
this is a test
" result = remove_inverted_text(text, 'en') if result != expected: print('text: ' + text) print('expected: ' + expected) print('result: ' + result) assert result == expected def run_all_tests(): base_dir = os.getcwd() print('Running tests...') update_default_themes_list(os.getcwd()) _test_source_contains_no_tabs() _translate_ontology(base_dir) _test_get_price_from_string() _test_post_variable_names() _test_config_param_names() _test_post_field_names('daemon.py', ['fields', 'actor_json']) _test_post_field_names('theme.py', ['config_json']) _test_post_field_names('inbox.py', ['queue_json', 'post_json_object', 'message_json', 'liked_post_json']) _test_checkbox_names() _test_thread_functions() _test_functions() _test_uninvert() _test_hashtag_maps() _test_combine_lines() _test_text_standardize() _test_dogwhistles() _test_remove_end_of_line() _test_translation_labels() _test_color_contrast_value(base_dir) _test_diff_content() _test_bold_reading() _test_published_to_local_timezone() _test_safe_webtext() _test_link_from_rss_item() _test_xml_podcast_dict(base_dir) _test_get_actor_from_in_reply_to() _test_valid_emoji_content() _test_add_cw_lists(base_dir) _test_word_similarity() _test_seconds_between_publish() _test_sign_and_verify() _test_danger_svg(base_dir) _test_can_replyto(base_dir) _test_date_conversions() _test_authorized_shared_items() _test_valid_password() _test_get_links_from_content() _test_set_actor_language() _test_limit_repeted_words() _test_word_lengths_limit() _test_switch_word(base_dir) _test_useragent_domain() _test_roles() _test_skills() _test_spoofed_geolocation() _test_remove_interactions() _test_extract_pgp_public_key() _test_emoji_images() _test_camel_case_split() _test_speaker_replace_link() _test_extract_text_fields_from_post() _test_markdown_to_html() _test_valid_hash_tag() _test_prepare_html_post_nick() _test_domain_handling() _test_mastoapi() _test_links_within_post(base_dir) _test_reply_to_public_post(base_dir) _test_mentioned_people(base_dir) _test_guess_tag_category() _test_valid_nick() _test_parse_newswire_feed_date() _test_first_paragraph_from_string() _test_newswire_tags() _test_hashtag_rules() _test_strip_html_tag() _test_replace_email_quote() _test_constant_time_string() _test_translations(base_dir) _test_valid_content_warning() _test_remove_id_ending() _test_json_post_allows_comment() _run_html_replace_quote_marks() _test_danger_css(base_dir) _test_danger_markup() _test_strip_html() _test_site_active() _test_jsonld() _test_remove_txt_formatting() _test_web_links() _test_recent_posts_cache() _test_theme() _test_save_load_json() _test_json_string() _test_get_status_number() _test_addemoji(base_dir) _test_actor_parsing() _test_httpsig(base_dir) _test_http_signed_get(base_dir) _test_http_sig_new('rsa-sha256', 'rsa-sha256') _test_httpsig_base_new(True, base_dir, 'rsa-sha256', 'rsa-sha256') _test_httpsig_base_new(False, base_dir, 'rsa-sha256', 'rsa-sha256') _test_cache() _test_threads() _test_create_person_account(base_dir) _test_authentication(base_dir) _test_followers_of_person(base_dir) _test_followers_on_domain(base_dir) _test_follows(base_dir) _test_group_followers(base_dir) time.sleep(2) print('Tests succeeded\n')