Coverage for distro_tracker/vendor/debian/management/commands/tracker_import_old_news.py: 78%

1# Copyright 2013-2016 The Distro Tracker Developers 

2# See the COPYRIGHT file at the top-level directory of this distribution and 

3# at https://deb.li/DTAuthors 

4# 

5# This file is part of Distro Tracker. It is subject to the license terms 

6# in the LICENSE file found in the top-level directory of this 

7# distribution and at https://deb.li/DTLicense. No part of Distro Tracker, 

8# including this file, may be copied, modified, propagated, or distributed 

9# except according to the terms contained in the LICENSE file. 

10"""Command to import news from a copy of the old PTS tree.""" 

11 

12import email 

13import os 

14from datetime import datetime 

15 

16from django.core.files.base import ContentFile 

17from django.core.management.base import BaseCommand, CommandError 

18from django.utils import timezone 

19 

20from distro_tracker.core.models import EmailNews, News, PackageName 

21 

22 

23class Command(BaseCommand): 

24 """ 

25 Import old PTS news. 

26 

27 The imported news' signature information is not automatically extracted. 

28 """ 

29 

30 def add_arguments(self, parser): 

31 parser.add_argument('rootdir') 

32 

33 def get_directories(self, root_directory): 

34 return [ 

35 os.path.join(root_directory, d) 

36 for d in os.listdir(root_directory) 

37 if os.path.isdir(os.path.join(root_directory, d)) 

38 ] 

39 

40 def write(self, message): 

41 if self.verbose: 41 ↛ 42line 41 didn't jump to line 42, because the condition on line 41 was never true

42 self.stdout.write(message) 

43 

44 def import_package_news(self, package_directory_name): 

45 package_name = os.path.basename(package_directory_name) 

46 self.write('Processing package {pkg}...'.format(pkg=package_name)) 

47 

48 try: 

49 package = PackageName.objects.get(name=package_name) 

50 except PackageName.DoesNotExist: 

51 self.write('Package does not exist. Skipping messages...') 

52 return 

53 

54 news_directory = os.path.join(package_directory_name, 'news') 

55 if not os.path.exists(news_directory): 55 ↛ 56line 55 didn't jump to line 56, because the condition on line 55 was never true

56 self.write("Package has no news directory.") 

57 return 

58 

59 email_news = [] 

60 for news_file in sorted(os.listdir(news_directory)): 

61 news_file_path = os.path.join(news_directory, news_file) 

62 

63 try: 

64 with open(news_file_path, 'rb') as f: 

65 if hasattr(email, 'message_from_binary_file'): 65 ↛ 68line 65 didn't jump to line 68, because the condition on line 65 was never false

66 msg = email.message_from_binary_file(f) 

67 else: 

68 msg = email.message_from_file(f) 

69 if 'Date' in msg: 69 ↛ 75line 69 didn't jump to line 75, because the condition on line 69 was never false

70 timestamp = email.utils.mktime_tz( 

71 email.utils.parsedate_tz(msg['Date'])) 

72 date = datetime.utcfromtimestamp(timestamp) 

73 date = timezone.make_aware(date, timezone.utc) 

74 else: 

75 date = timezone.now() 

76 

77 news_kwargs = EmailNews.get_email_news_parameters(msg) 

78 content = news_kwargs.pop('file_content') 

79 news_kwargs['news_file'] = ContentFile(content, 

80 name='news-file') 

81 

82 email_news.append(News( 

83 package=package, 

84 datetime_created=date, 

85 **news_kwargs)) 

86 except Exception: 

87 import traceback 

88 traceback.print_exc() 

89 self.write('Problem importing news {}'.format(news_file_path)) 

90 

91 self.write("All news for the package processed. " 

92 "Bulk creating the instances.") 

93 News.objects.bulk_create(email_news) 

94 

95 self.write('Complete.') 

96 

97 def import_all_news(self, root_directory): 

98 for hash_directory in self.get_directories(root_directory): 

99 for package_directory in self.get_directories(hash_directory): 

100 self.import_package_news(package_directory) 

101 

102 def handle(self, *args, **kwargs): 

103 if 'rootdir' not in kwargs or not kwargs['rootdir']: 103 ↛ 104line 103 didn't jump to line 104, because the condition on line 103 was never true

104 raise CommandError("Root directory of old news not provided") 

105 self.verbose = int(kwargs.get('verbosity', 1)) > 1 

106 

107 # Hack to be able to set the date created field to something else 

108 # than now. 

109 EmailNews._meta.get_field('datetime_created').auto_now_add = False 

110 

111 self.import_all_news(kwargs['rootdir']) 

112 

113 EmailNews._meta.get_field('datetime_created').auto_now_add = True