Created
June 29, 2017 09:17
-
-
Save shailesh/3a04f26333111d78ba73c5c7b694ef27 to your computer and use it in GitHub Desktop.
I have some friends records in a text file (friends.json) -- one friend per line, JSON-encoded. I want to invite any friends within 100km of our Bangalore office which is almost like our home (GPS coordinates 12.9611159,77.6362214) for some food and drinks on us. Write a program that will read the full list of friends and output the names and us…
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
{"latitude": "12.986375", "user_id": 12, "name": "Chris", "longitude": "77.043701"} | |
{"latitude": "11.92893", "user_id": 1, "name": "Alice", "longitude": "78.27699"} | |
{"latitude": "11.8856167", "user_id": 2, "name": "Ian", "longitude": "78.4240911"} | |
{"latitude": "12.3191841", "user_id": 3, "name": "Jack", "longitude": "78.5072391"} | |
{"latitude": "13.807778", "user_id": 28, "name": "Charlie", "longitude": "76.714444"} | |
{"latitude": "13.4692815", "user_id": 7, "name": "Frank", "longitude": "-9.436036"} | |
{"latitude": "14.0894797", "user_id": 8, "name": "Eoin", "longitude": "77.18671"} | |
{"latitude": "13.038056", "user_id": 26, "name": "Stephen", "longitude": "76.613889"} | |
{"latitude": "14.1225", "user_id": 27, "name": "Enid", "longitude": "78.143333"} | |
{"latitude": "13.1229599", "user_id": 6, "name": "Theresa", "longitude": "77.2701202"} | |
{"latitude": "12.2559432", "user_id": 9, "name": "Jack", "longitude": "76.1048927"} | |
{"latitude": "12.240382", "user_id": 10, "name": "Georgina", "longitude": "77.972413"} | |
{"latitude": "13.2411022", "user_id": 4, "name": "Ian", "longitude": "77.238335"} | |
{"latitude": "13.1302756", "user_id": 5, "name": "Nora", "longitude": "77.2397222"} | |
{"latitude": "13.008769", "user_id": 11, "name": "Richard", "longitude": "77.1056711"} | |
{"latitude": "13.1489345", "user_id": 31, "name": "Alan", "longitude": "77.8422408"} | |
{"latitude": "13", "user_id": 13, "name": "Olive", "longitude": "76"} | |
{"latitude": "11.999447", "user_id": 14, "name": "Helen", "longitude": "-9.742744"} | |
{"latitude": "12.966", "user_id": 15, "name": "Michael", "longitude": "77.463"} | |
{"latitude": "12.366037", "user_id": 16, "name": "Ian", "longitude": "78.179118"} | |
{"latitude": "14.180238", "user_id": 17, "name": "Patricia", "longitude": "-5.920898"} | |
{"latitude": "13.0033946", "user_id": 39, "name": "Lisa", "longitude": "77.3877505"} | |
{"latitude": "12.228056", "user_id": 18, "name": "Bob", "longitude": "76.915833"} | |
{"latitude": "14.133333", "user_id": 24, "name": "Rose", "longitude": "77.433333"} | |
{"latitude": "55.033", "user_id": 19, "name": "Enid", "longitude": "78.112"} | |
{"latitude": "13.121111", "user_id": 20, "name": "Enid", "longitude": "-9.831111"} | |
{"latitude": "11.802", "user_id": 21, "name": "David", "longitude": "-9.442"} | |
{"latitude": "14.374208", "user_id": 22, "name": "Charlie", "longitude": "78.371639"} | |
{"latitude": "13.74412", "user_id": 29, "name": "Oliver", "longitude": "76.11167"} | |
{"latitude": "13.761389", "user_id": 30, "name": "Nick", "longitude": "76.2875"} | |
{"latitude": "14.080556", "user_id": 23, "name": "Eoin", "longitude": "77.361944"} | |
{"latitude": "12.833502", "user_id": 25, "name": "David", "longitude": "78.122366"} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from concurrent.futures import ProcessPoolExecutor | |
from functools import partial | |
from operator import itemgetter | |
import json | |
import os.path | |
import math | |
import time | |
import invite_friends_config as config | |
from utils import is_valid_location, is_valid_friends_data, is_valid_config, LOCATION, SOURCE_LOCATION | |
from log_conf import get_logger | |
LOGGER = get_logger(__name__) | |
RADIUS = 6371 | |
SMALL_FILE_SIZE = 8 | |
FILE_INPUT_BUFFER_SIZE = SMALL_FILE_SIZE * 1024 * 1024 | |
SHOULD_VALIDATE_INDIVIDUAL_DATA = True | |
DISTANCE_RANGE = 100 | |
GLOBAL_RESULT = [] | |
bad_inputs = [] | |
def calculate_distance(location_destination, location_source=SOURCE_LOCATION): | |
if location_source != SOURCE_LOCATION and not is_valid_location(location_source): | |
raise Exception | |
if not is_valid_location(location_destination): | |
raise Exception | |
latitude_source = float(location_source.latitude) | |
longitude_source = float(location_source.longitude) | |
latitude_destination = float(location_destination.latitude) | |
longitude_destination = float(location_destination.longitude) | |
phi_1 = math.radians(latitude_source) | |
phi_2 = math.radians(latitude_destination) | |
delta_lambda = math.radians(longitude_destination - longitude_source) | |
delta_sigma = math.acos((math.sin(phi_1) * math.sin(phi_2)) + | |
(math.cos(phi_1) * math.cos(phi_2) * math.cos(delta_lambda))) | |
result = RADIUS * delta_sigma | |
return result | |
def process_friends_data_worker(data, testing=False): | |
if testing: | |
bad_data_testing = list() | |
good_result_testing = list() | |
good_data_testing = list() | |
test_data_length = len(data) | |
distance = 101.0 | |
for friend_item in data: | |
json_decoded_friends_data = json.loads(friend_item) | |
LOGGER.debug('CHECK BUG LINE 1 : json_decoded_friends_data : type= %s and\nValue : %s ' % (str(type(json_decoded_friends_data)), json_decoded_friends_data)) | |
if SHOULD_VALIDATE_INDIVIDUAL_DATA and not is_valid_friends_data(json_decoded_friends_data): | |
LOGGER.debug('Not Valid Friends Data : data = %s ' % json_decoded_friends_data) | |
if not testing: | |
LOGGER.debug('bad_inputs data appending : %s ' % json_decoded_friends_data) | |
bad_inputs.append(json_decoded_friends_data) | |
LOGGER.debug('bad_inputs lenthg : %d ' % len(bad_inputs)) | |
else: | |
LOGGER.debug('bad_testing_data appending : %s ' % json_decoded_friends_data) | |
bad_data_testing.append(json_decoded_friends_data) | |
LOGGER.debug('bad_testing_data lenthg : %d ' % len(bad_data_testing)) | |
LOGGER.debug('continue with rest of loop') | |
continue | |
LOGGER.debug('THIS LINE SHOULD NOT BE ADDED - IF ADDED THEN SOMETHING WRONG WITH CODE') | |
friends_location_coordinates = LOCATION(json_decoded_friends_data['latitude'], json_decoded_friends_data['longitude']) | |
try: | |
distance = calculate_distance(location_destination=friends_location_coordinates) | |
except: | |
if not testing: | |
LOGGER.debug('bad_inputs data appending : %s ' % json_decoded_friends_data) | |
bad_inputs.append(json_decoded_friends_data) | |
LOGGER.debug('bad_inputs lenthg : %d ' % len(bad_inputs)) | |
else: | |
LOGGER.debug('bad_testing_data appending : %s ' % json_decoded_friends_data) | |
bad_data_testing.append(json_decoded_friends_data) | |
LOGGER.debug('bad_testing_data lenthg : %d ' % len(bad_data_testing)) | |
LOGGER.debug('Continue Statement since bad_data') | |
continue | |
LOGGER.debug('THIS LINE SHOULD NOT BE PRINTED, IF IT DOES, THEN SOMETHIN IS WRONG IN CODE') | |
if distance < DISTANCE_RANGE: | |
LOGGER.debug('json_decoded_friends_data : type= %s and\nValue : %s ' % (str(type(json_decoded_friends_data)), json_decoded_friends_data)) | |
LOGGER.debug('distance qualified : %f ' % distance) | |
if not testing: | |
LOGGER.debug('Adding Good Resuly appending : %s ' % json_decoded_friends_data) | |
GLOBAL_RESULT.append(json_decoded_friends_data) | |
LOGGER.debug('GLOBAL_RESULT lenthg : %d ' % len(GLOBAL_RESULT)) | |
else: | |
LOGGER.debug('Adding Testing Good Result appending : %s ' % json_decoded_friends_data) | |
good_result_testing.append(json_decoded_friends_data) | |
LOGGER.debug('good_result_testing lenthg : %d ' % len(good_result_testing)) | |
LOGGER.debug('CHECK BUG LINE 2') | |
else: | |
if testing: | |
good_data_testing.append(json_decoded_friends_data) | |
LOGGER.debug('GOOD_RESULT TILL NOW length : %d ' % len(GLOBAL_RESULT)) | |
LOGGER.debug('bad_inputs TIll Now length : %d ' % len(bad_inputs)) | |
if testing: | |
if len(bad_data_testing) + len(good_result_testing) + len(good_data_testing) != test_data_length: | |
return False | |
return True | |
def _dummy_process_data_worker_function(data): | |
if data is not None and len(data) > 0: | |
return True | |
raise Exception | |
def read_friend_list(source_path, asynchronous, testing=False): | |
file_size = None | |
executor_futures_result = None | |
try: | |
file_size = os.path.getsize(source_path) / float(1024 * 1024) | |
except OSError: | |
LOGGER.error('File does not exist or permission is denied at path: %s ' % source_path) | |
raise | |
friends_data = list() | |
if file_size <= SMALL_FILE_SIZE: | |
LOGGER.info('Input Friend List file size is small : size in MB = %d ' % file_size) | |
with open(source_path, 'rb') as input_file: | |
friends_data = input_file.readlines() | |
if not testing: | |
process_friends_data_worker(friends_data) | |
else: | |
LOGGER.info('Testing read_friend_list functionality, hence not passing the opened file for processing') | |
else: | |
LOGGER.info('Input Friend List file size is large : size in MB = %s ' % str(file_size)) | |
with open(source_path, 'rb') as input_file: | |
LOGGER.debug('Input File Opened as input_file') | |
if asynchronous is True: | |
LOGGER.debug('asynchronous is True') | |
with ProcessPoolExecutor() as executor: | |
partial_readlines = partial(input_file.readlines, FILE_INPUT_BUFFER_SIZE) | |
iterator_for_executor_map = iter(partial_readlines, []) | |
if not testing: | |
try: | |
executor.map(process_friends_data_worker, iterator_for_executor_map, chunksize=1) | |
except: | |
return | |
else: | |
try: | |
executor.map(_dummy_process_data_worker_function, iterator_for_executor_map, chunksize=1) | |
except: | |
return | |
return executor_futures_result | |
else: | |
LOGGER.debug('asynchronous is False') | |
while True: | |
LOGGER.debug('Starting While Loop') | |
friends_data = input_file.readlines(FILE_INPUT_BUFFER_SIZE) | |
if not friends_data: | |
LOGGER.debug('Friend data is None') | |
LOGGER.debug('Breaking from Loop Now') | |
break | |
LOGGER.debug('THIS LINE SHOULD NOT BE PRINTED _ ELSE SOMETHING WRONG WITH CODE') | |
else: | |
LOGGER.debug('Friend data is not None - calling process_friends_data_worker') | |
if not testing: | |
LOGGER.debug('Testing = False --> Friend data is not None - calling process_friends_data_worker') | |
process_friends_data_worker(friends_data) | |
if testing: | |
LOGGER.debug('Testing is True, now return True') | |
return True | |
def print_bad_inputs(inputs=None): | |
if inputs is None: | |
inputs = bad_inputs | |
count = len(inputs) | |
if count < 0: | |
LOGGER.error('Found %d bad_inputs ' % count) | |
print 'Bad Inputs : \n' | |
for i in xrange(count): | |
print str(i + 1) + '. ', inputs[i] | |
else: | |
print '' | |
def print_invitation_list(invitation_list=None): | |
if invitation_list is None: | |
invitation_list = GLOBAL_RESULT | |
if invitation_list == []: | |
LOGGER.info('Found 0 friends nearby to send invitations') | |
print 'Found 0 friends nearby to send invitations' | |
else: | |
invitation_list.sort(key=itemgetter('user_id')) | |
count = len(invitation_list) | |
LOGGER.info('Found %d qualified friends ' % count) | |
print 'The names and user ids of matching friends (within 100km), sorted by user id:\n' | |
for i in xrange(count): | |
print str(i + 1) + '. ' + 'Id: ', invitation_list[i]['user_id'], ' Name: ' + invitation_list[i]['name'] | |
def invite_friends(friend_list_source, source_path=None, asynchronous=False): | |
if friend_list_source == 'file': | |
try: | |
start_time = time.time() | |
LOGGER.debug('Call read_friend_list()') | |
read_friend_list(source_path, asynchronous) | |
LOGGER.debug('Returned from read_friend_list()') | |
LOGGER.debug('GOOD_RESULT FINAL length : %d ' % len(GLOBAL_RESULT)) | |
LOGGER.debug('bad_inputs FINAL length : %d ' % len(bad_inputs)) | |
reading_time = time.time() - start_time | |
LOGGER.info('Invitatoin Generation Time: %f seconds ' % reading_time) | |
print 'Invitatoin Generation Time: %f seconds ' % reading_time | |
except: | |
return | |
print_bad_inputs() | |
print_invitation_list() | |
else: | |
print '\nThese advanced features will be implemented soon\nKeep following\n\n' | |
raise NotImplementedError | |
def main(): | |
config_data = config.inputs | |
if not is_valid_config(config_data): | |
return | |
friend_list_source = config_data['friend_list_source'] | |
source_path = config_data['source_path'] | |
asynchronous = config_data['async'] | |
invite_friends(friend_list_source, source_path, asynchronous) | |
if __name__ == "__main__": | |
main() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
inputs = { | |
'friend_list_source': 'file', | |
'source_path': 'friends.json', | |
'async': False | |
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import logging | |
DEFAULT_FORMAT = '[%(levelname)s]: [%(asctime)s] [%(name)s] [%(module)s:%(lineno)d] - %(message)s' | |
DEFAULT_DATE_FORMAT = '%d-%m-%y %H:%M:%S' | |
DEFAULT_LOG_FILE = 'invite_friends.log' | |
DEFAULT_FILE_LOGLEVEL = logging.INFO | |
DEFAULT_STREAM_LOGLEVEL = logging.ERROR | |
DEFAULT_LOGGER_LOGLEVEL = logging.INFO | |
def get_logger(module_name): | |
logging.basicConfig(level=DEFAULT_FILE_LOGLEVEL, filename=DEFAULT_LOG_FILE, format=DEFAULT_FORMAT, datefmt=DEFAULT_DATE_FORMAT) | |
stream_handler = logging.StreamHandler() | |
stream_handler.setLevel(DEFAULT_STREAM_LOGLEVEL) | |
formatter = logging.Formatter(DEFAULT_FORMAT, DEFAULT_DATE_FORMAT) | |
stream_handler.setFormatter(formatter) | |
stream_logger = logging.getLogger(module_name) | |
stream_logger.setLevel(DEFAULT_LOGGER_LOGLEVEL) | |
stream_logger.addHandler(stream_handler) | |
return stream_logger |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import pytest | |
import random | |
import json | |
import time | |
from invite_friends import calculate_distance, process_friends_data_worker, read_friend_list,\ | |
print_bad_inputs, print_invitation_list, invite_friends | |
from utils import LOCATION | |
def setup_module(module): | |
print "\nsetup_module module: %s \n\n" % module.__name__ | |
try: | |
create_large_input_file('friends_large.json', 1000000) | |
except: | |
print 'unable to create large input file' | |
def teardown_module(module): | |
print "\n\nteardown_module module: %s " % module.__name__ | |
def test_calculate_distance_with_valid_inputs(): | |
destination = LOCATION('12.986375', '77.043701') | |
assert calculate_distance(destination) == 64.26480291995638 | |
destination = LOCATION('80.123456', '-175.12345') | |
assert calculate_distance(destination) == 8909.980105033259 | |
destination = LOCATION('-85.654321', '175.654321') | |
assert calculate_distance(destination) == 11511.948749243722 | |
def test_calculate_distance_with_invalid_inputs(): | |
with pytest.raises(Exception): | |
destination = LOCATION('112.986375', '77.043701') | |
assert calculate_distance(destination) | |
with pytest.raises(Exception): | |
destination = LOCATION('80.1324242', '-181.000000') | |
assert calculate_distance(destination) | |
with pytest.raises(Exception): | |
destination = LOCATION('12,134134', '77.123456') | |
assert calculate_distance(destination) | |
with pytest.raises(Exception): | |
destination = LOCATION(12.987654, '77.123456') | |
assert calculate_distance(destination) | |
with pytest.raises(Exception): | |
destination = LOCATION('12.975310', '77.8ac765') | |
assert calculate_distance(destination) | |
with pytest.raises(Exception): | |
destination = LOCATION('a12.457984', '77.87654b') | |
assert calculate_distance(destination) | |
with pytest.raises(Exception): | |
destination = LOCATION(('12.986375', '77.043701')) | |
assert calculate_distance(destination) | |
with pytest.raises(Exception): | |
assert calculate_distance() | |
def create_large_input_file(filename, num_of_entries): | |
def _lat(): | |
return "{:.6f}".format(random.uniform(-90, 90)) | |
def _lon(): | |
return "{:.6f}".format(random.uniform(-180, 180)) | |
def _user_name(): | |
return ''.join([random.choice('abcdefgjhijklmnopqrstuvwzyz') for _ in xrange(random.randrange(5, 15))]) +\ | |
''.join([random.choice('abcdefgjhijklmnopqrstuvwzyz') | |
for _ in xrange(random.randrange(5, 15))]) | |
def _create_item(user_id): | |
return {'latitude': _lat(), 'user_id': user_id, 'name': _user_name(), 'longitude': _lon()} | |
with open(filename, 'w') as output_file: | |
for i in xrange(num_of_entries): | |
json.dump(_create_item(i), output_file) | |
output_file.write('\n') | |
def test_read_friend_list_with_valid_source_path(): | |
s = time.time() | |
assert read_friend_list(source_path='friends.json', asynchronous=False, testing=True) is True | |
print 'Done' | |
print 'Reading took : %s seconds' % str(time.time() - s) | |
s = time.time() | |
assert read_friend_list(source_path='friends.json', asynchronous=True, testing=True) is True | |
print 'Done' | |
print 'Reading took : %s seconds' % str(time.time() - s) | |
s = time.time() | |
assert read_friend_list(source_path='friends_large.json', | |
asynchronous=False, testing=True) is True | |
print 'Done' | |
print 'Reading took : %s seconds' % str(time.time() - s) | |
s = time.time() | |
assert read_friend_list(source_path='friends_large.json', | |
asynchronous=True, testing=True) is None | |
print 'Done' | |
print 'Reading took : %s seconds' % str(time.time() - s) | |
def test_read_friend_list_with_invalid_source_path(): | |
with pytest.raises(Exception): | |
assert read_friend_list(source_path='friens.json', asynchronous=False, testing=True) | |
with pytest.raises(Exception): | |
assert read_friend_list(source_path='/usr/local/etc/friends.json', | |
asynchronous=True, testing=True) | |
def test_process_friends_data_worker_with_valid_data(): | |
with open('friends.json', 'rb') as input_file: | |
data = input_file.readlines(8192) | |
assert process_friends_data_worker(data, testing=True) is True | |
def test_print_bad_inputs(): | |
inputs = [ | |
{"latitude": "12.240382", "user_id": "10", "name": "Georgina", "longitude": "77.972413"}, | |
{"latitude": "12.240382", "user_id": 10, "name": "Georgina", "longitude": 77.972413} | |
] | |
assert print_bad_inputs(inputs) is None | |
def test_print_invitation_list(): | |
invitation_list = [ | |
{"latitude": "12.240382", "user_id": 10, "name": "Georgina", "longitude": "77.972413"}, | |
{"latitude": "13.2411022", "user_id": 4, "name": "Ian", "longitude": "77.238335"}, | |
{"latitude": "13.1302756", "user_id": 5, "name": "Nora", "longitude": "77.2397222"} | |
] | |
assert print_invitation_list(invitation_list) is None | |
def test_invite_friends_with_source_as_file(): | |
assert invite_friends(friend_list_source='file', source_path='friends.json', asynchronous=False) is None | |
def test_invite_friends_with_source_not_as_file(): | |
with pytest.raises(NotImplementedError): | |
assert invite_friends(friend_list_source='mongodb', source_path=None, asynchronous=False) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from utils import is_valid_coordinate_format, is_valid_coordinates_value, is_valid_location,\ | |
is_valid_friends_data, is_valid_config, LOCATION | |
def setup_module(module): | |
print "\nsetup_module module: %s \n\n" % module.__name__ | |
def teardown_module(module): | |
print "\n\nteardown_module module: %s " % module.__name__ | |
def test_is_valid_coordinate_format_with_valid_coordinate(): | |
assert is_valid_coordinate_format(' 12.986375') is True | |
assert is_valid_coordinate_format('-152.986375 ') is True | |
assert is_valid_coordinate_format(' -152.986375 ') is True | |
def test_is_valid_coordinate_format_with_Invalid_coordinate(): | |
assert is_valid_coordinate_format('a12.986375') is False | |
assert is_valid_coordinate_format('-152986375') is False | |
assert is_valid_coordinate_format('12.986375a') is False | |
assert is_valid_coordinate_format('-1529 86375') is False | |
assert is_valid_coordinate_format('15,86375') is False | |
assert is_valid_coordinate_format('ab.cdefghi') is False | |
assert is_valid_coordinate_format() is False | |
def test_is_valid_coordinates_value_with_valid_coordinates(): | |
assert is_valid_coordinates_value('12.986375', '77.043701') is True | |
assert is_valid_coordinates_value('80.123456', '-175.12345') is True | |
assert is_valid_coordinates_value('-85.654321', '175.654321') is True | |
def test_is_valid_coordinates_value_with_Invalid_coordinates(): | |
assert is_valid_coordinates_value('112.986375', '77.043701') is False | |
assert is_valid_coordinates_value('80.1324242', '-181.000000') is False | |
assert is_valid_coordinates_value('-112.134134', '77.123456') is False | |
assert is_valid_coordinates_value('77.123456') is False | |
assert is_valid_coordinates_value() is False | |
def test_is_valid_location_with_valid_location(): | |
destination = LOCATION('12.986375', '77.043701') | |
assert is_valid_location(destination) is True | |
destination = LOCATION('80.123456', '-175.12345') | |
assert is_valid_location(destination) is True | |
destination = LOCATION('-85.654321', '175.654321') | |
assert is_valid_location(destination) is True | |
def test_is_valid_location_with_Invalid_location(): | |
destination = ('12.986375', '77.043701') | |
assert is_valid_location(destination) is False | |
destination = LOCATION('80.1324242', '-181.000000') | |
assert is_valid_location(destination) is False | |
destination = LOCATION('12,134134', '77.123456') | |
assert is_valid_location(destination) is False | |
destination = LOCATION(12.987654, '77.123456') | |
assert is_valid_location(destination) is False | |
destination = LOCATION('12.975310', '77.8ac765') | |
assert is_valid_location(destination) is False | |
destination = LOCATION('a12.457984', '77.87654b') | |
assert is_valid_location(destination) is False | |
assert is_valid_location() is False | |
def test_is_valid_friends_data_with_valid_data(): | |
data = {"latitude": "12.986375", "user_id": 12, "name": "Chris", "longitude": "77.043701"} | |
assert is_valid_friends_data(data) is True | |
def test_is_valid_friends_data_with_Invalid_data(): | |
data = [{"latitud": "12.986375", "user_id": 12, "name": "Chris", "longitude": "77.043701"}] | |
assert is_valid_friends_data(data) is False | |
data = data[0] | |
assert is_valid_friends_data(data) is False | |
data = {"latitude": 12.986375, "user_id": 12, "name": "Chris", "longitude": "77.043701"} | |
assert is_valid_friends_data(data) is False | |
data = {"latitude": "12.986375", "id": 12, "name": "Chris", "longitude": "77.043701"} | |
assert is_valid_friends_data(data) is False | |
data = {"latitud": "12.986375", "user_id": "12afd3", "name": "Chris", "longitude": "77.043701"} | |
assert is_valid_friends_data(data) is False | |
data = {"latitud": "12.986375", "user_id": 12, "username": "Chris", "longitude": "77.043701"} | |
assert is_valid_friends_data(data) is False | |
data = {"latitude": 12.986375, "user_id": 12, "name": ["Chris", "harris"], "longitude": "77.043701"} | |
assert is_valid_friends_data(data) is False | |
data = {"latitude": "12.986375", "id": 12, "name": "Chris", "lon": "77.043701"} | |
assert is_valid_friends_data(data) is False | |
data = {"latitude": "12.986375", "id": 12, "name": "Chris", "longitude": 77.043701} | |
assert is_valid_friends_data(data) is False | |
assert is_valid_friends_data() is False | |
def test_is_valid_config_with_good_configurations(): | |
good_config = { | |
'friend_list_source': 'file', | |
'source_path': 'friends.json', | |
'async': True | |
} | |
assert is_valid_config(good_config) is True | |
def test_is_valid_config_with_bad_configurations(): | |
bad_config = [{ | |
'source': 'file', | |
'source_path': 'friends.json', | |
'async': True | |
}] | |
assert is_valid_config(bad_config) is False | |
bad_config = bad_config[0] | |
assert is_valid_config(bad_config) is False | |
del bad_config['source'] | |
bad_config['friend_list_source'] = 1 | |
assert is_valid_config(bad_config) is False | |
bad_config['friend_list_source'] = 'invalid_source' | |
assert is_valid_config(bad_config) is False | |
bad_config['friend_list_source'] = 'file' | |
bad_config['source_path'] = ('path/to/source') | |
assert is_valid_config(bad_config) is False | |
del bad_config['source_path'] | |
assert is_valid_config(bad_config) is False | |
bad_config['source_path'] = 'friends.json' | |
bad_config['async'] = 1 | |
assert is_valid_config(bad_config) is False | |
del bad_config['async'] | |
assert is_valid_config(bad_config) is False | |
bad_config['asynchronous'] = True | |
assert is_valid_config(bad_config) is False | |
assert is_valid_config() is False |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import os.path | |
from collections import namedtuple | |
from log_conf import get_logger | |
LOGGER = get_logger(__name__) | |
LOCATION = namedtuple('LOCATION', 'latitude, longitude') | |
SOURCE_LOCATION = LOCATION('12.9611159', '77.6362214') | |
FRIEND_LIST_SOURCES = ('file',) | |
def is_valid_coordinate_format(coordinate_point=None): | |
if not coordinate_point: | |
return False | |
coordinate_point = coordinate_point.strip() | |
if coordinate_point.count('.') == 1: | |
number_before_decimal, number_after_decimal = coordinate_point.split('.') | |
if number_before_decimal.startswith('-'): | |
number_before_decimal = number_before_decimal[1:] | |
if number_before_decimal.isdigit() and number_after_decimal.isdigit(): | |
return True | |
else: | |
return False | |
else: | |
return False | |
def is_valid_coordinates_value(latitude=None, longitude=None): | |
if not latitude or not longitude: | |
return False | |
latitude = float(latitude) | |
longitude = float(longitude) | |
if latitude < -90 or latitude > 90 or longitude < -180 or longitude > 180: | |
return False | |
return True | |
def is_valid_location(location=None): | |
if not location: | |
return False | |
if not isinstance(location, LOCATION): | |
LOGGER.error( | |
'Invalid Location type: location should be of type LOCATION, given location is of type : %s ' % str(type(location))) | |
return False | |
latitude = location.latitude | |
longitude = location.longitude | |
if (not isinstance(latitude, str) and not isinstance(latitude, unicode)) or (not isinstance(longitude, str) and not isinstance(longitude, unicode)): | |
LOGGER.error('Invalid Location: latitude and longitude should both be of type str or unicode, given latitude type = %s and longitude type = %s ' % (str(type(latitude)), str(type(longitude)))) | |
return False | |
if not is_valid_coordinate_format(latitude): | |
LOGGER.error( | |
'Invalid Latitude coordinate: latitude should be in degrees - Latitude: %s ' % str(latitude)) | |
return False | |
if not is_valid_coordinate_format(longitude): | |
LOGGER.error( | |
'Invalid Longitude coordinate: longitude should be in degrees - Longitude: %s ' % str(longitude)) | |
return False | |
if not is_valid_coordinates_value(latitude, longitude): | |
LOGGER.error( | |
'Invalid coordinate values: Should be -90<=latitude<=90 and -180<=longitude<=180 Given are : %s, %s ' % (latitude, longitude)) | |
return False | |
return True | |
def is_valid_friends_data(data=None): | |
if not data: | |
return False | |
if not isinstance(data, dict): | |
# logs the error regarding invalid friend data | |
LOGGER.error('Bad Data: Data not according to required structure: Data should be in dictionary form') | |
return False | |
if 'latitude' not in data or 'longitude' not in data or 'name' not in data or 'user_id' not in data: | |
# logs the error regarding invalid inputs keys | |
LOGGER.error( | |
'Invalid Data keys. Data should have all the 4 keys, [longitude], [latitude], [name] and [id]') | |
return False | |
if (not isinstance(data['name'], str) and not isinstance(data['name'], unicode)) or not isinstance(data['user_id'], int): | |
# logs the error regarding invalid friend data value types | |
LOGGER.error('Bad Data: Data not according to required structure: Name should be of type str or unicode and id of type int.\ | |
Given data : type(data[name]) : %s and type(data[id]) : %s ' % (str(type(data['name'])), str(type(data['user_id'])))) | |
return False | |
if (not isinstance(data['latitude'], str) and not isinstance(data['latitude'], unicode)) or (not isinstance(data['longitude'], str) and not isinstance(data['longitude'], unicode)): | |
# logs the error regarding invalid friend data value types | |
LOGGER.error('Bad Data: Data not according to required structure: Latitude and Longitude should be of type str or unicode.\ | |
Given data : type(data[latitude]) : %s and type(data[longitude]) : %s ' % (str(type(data['latitude'])), str(type(data['longitude'])))) | |
return False | |
return True | |
def is_valid_config(config_data=None): | |
if not config_data: | |
return False | |
if not isinstance(config_data, dict): | |
LOGGER.error('Inputs is not valid: Inputs in settings should be a dictionary object') | |
return False | |
if 'friend_list_source' not in config_data or 'source_path' not in config_data or 'async' not in config_data: | |
LOGGER.error( | |
'Invalid Inputs keys. Settings\' inputs dictionary should have all the 3 keys, [friend_list_source], [source_path] and [async]') | |
return False | |
friend_list_source = config_data['friend_list_source'] | |
source_path = config_data['source_path'] | |
asynchronous = config_data['async'] | |
if not isinstance(friend_list_source, str): | |
LOGGER.error('Invalid source type. Source should be of type str.') | |
return False | |
if friend_list_source not in FRIEND_LIST_SOURCES: | |
LOGGER.error('Invalid source value. Source Value should be from %s ' % str(FRIEND_LIST_SOURCES)) | |
if (not isinstance(source_path, str) and friend_list_source == 'file') or\ | |
(source_path is not None and friend_list_source != 'file'): | |
LOGGER.error( | |
'Invalid Source Path type. Source Path should be of type str when source if file or None if source is not file. ') | |
return False | |
if not isinstance(asynchronous, bool): | |
LOGGER.error('Invalid async type. Async should of type bool, i.e, eiher True or False.') | |
return False | |
if friend_list_source == 'file' and not os.path.exists(source_path): | |
LOGGER.error('Invalid source path value. source path should a proper directory path') | |
return False | |
if not is_valid_location(SOURCE_LOCATION): | |
return False | |
return True |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment