19684186419

Committed 25 Nov 2025 09:10PM UTC coverage: 79.751% (+3.9%) from 75.877%

Build # 19684186419

Build Type

Pull #18

github

Committed by

web-flow

Commit Message

Merge 333b4230e into cd57f2a02

Pull Request Pull Request #18: more tests

Run Details

106 of 115 new or added lines in 3 files covered. (92.17%)

37 existing lines in 4 files now uncovered.

1024 of 1284 relevant lines covered (79.75%)

0.8 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

72.53

/training_provisioner/csv/data.py

# Copyright 2025 UW-IT, University of Washington
# SPDX-License-Identifier: Apache-2.0


from django.conf import settings
from django.core.files.storage import default_storage
from training_provisioner.csv.format import (
    UserHeader, AdminHeader, TermHeader, CourseHeader,
    SectionHeader, EnrollmentHeader, UserCSV,
    AdminCSV, TermCSV, CourseCSV, SectionCSV, EnrollmentCSV)
from datetime import datetime
from logging import getLogger
import os

logger = getLogger(__name__)


class Collector(object):
    def __init__(self):
        self._init_data()

    def _init_data(self):
        self.accounts = []
        self.admins = []
        self.terms = {}
        self.courses = {}
        self.sections = {}
        self.enrollments = []
        self.enrollment_keys = {}
        self.users = {}
        self.headers = {
            'users': UserHeader(),
            'admins': AdminHeader(),
            'terms': TermHeader(),
            'courses': CourseHeader(),
            'sections': SectionHeader(),
            'enrollments': EnrollmentHeader(),
        }

    def add(self, formatter):
        """
        Add the passed csv formatter object based on type, returns True if
        the formatter is added, False otherwise.
        """
        if isinstance(formatter, UserCSV):
            return self._add_user(formatter)
        elif isinstance(formatter, EnrollmentCSV):
            return self._add_enrollment(formatter)
        elif isinstance(formatter, AdminCSV):
            return self._add_admin(formatter)
        elif isinstance(formatter, TermCSV):
            return self._add_term(formatter)
        elif isinstance(formatter, CourseCSV):
            return self._add_course(formatter)
        elif isinstance(formatter, SectionCSV):
            return self._add_section(formatter)
        else:
            raise TypeError(
                'Unknown CSVFormat class: {}'.format(type(formatter)))

    def _add_admin(self, formatter):
        self.admins.append(formatter)
        return True

    def _add_user(self, formatter):
        if formatter.key not in self.users:
            self.users[formatter.key] = formatter
            return True
        return False

    def _add_term(self, formatter):
        if formatter.key not in self.terms:
            self.terms[formatter.key] = formatter
            return True
        return False

    def _add_course(self, formatter):
        if formatter.key not in self.courses:
            self.courses[formatter.key] = formatter
            return True
        return False

    def _add_section(self, formatter):
        if formatter.key not in self.sections:
            self.sections[formatter.key] = formatter
            return True
        return False

    def _add_enrollment(self, formatter):
        if formatter.key not in self.enrollment_keys:
            self.enrollment_keys[formatter.key] = True
            self.enrollments.append(formatter)
            return True
        return False

    def has_data(self):
        """
        Returns True if the collector contains data, False otherwise.
        """
        for csv_type in self.headers:
            if len(getattr(self, csv_type)):
                return True
        return False

    def write_files(self):
        """
        Writes all csv files. Returns a path to the csv files, or None
        if no data was written.
        """
        filepath = None
        if self.has_data():
            filepath = datetime.now().strftime('%Y/%m/%d/%H%M%S-%f')
            for csv_type in self.headers:
                try:
                    data = list(getattr(self, csv_type).values())
                    data.sort()
                except AttributeError:
                    data = getattr(self, csv_type)

                if len(data):
                    os.makedirs(os.path.join(
                        settings.MEDIA_ROOT, filepath), exist_ok=True)
                    filename = os.path.join(filepath, csv_type + '.csv')
                    with default_storage.open(filename, mode='w') as f:
                        headers = self.headers[csv_type]
                        f.write(str(headers))
                        for line in data:
                            f.write(str(line))

            self._init_data()

        if getattr(settings, 'TRAINING_IMPORT_CSV_DEBUG', False):
            logger.debug('CSV PATH: {}'.format(filepath))
            return None
        else:
            return filepath

1	# Copyright 2025 UW-IT, University of Washington
2	# SPDX-License-Identifier: Apache-2.0
3
4
5	from django.conf import settings	1✔
6	from django.core.files.storage import default_storage	1✔
7	from training_provisioner.csv.format import (	1✔
8	UserHeader, AdminHeader, TermHeader, CourseHeader,
9	SectionHeader, EnrollmentHeader, UserCSV,
10	AdminCSV, TermCSV, CourseCSV, SectionCSV, EnrollmentCSV)
11	from datetime import datetime	1✔
12	from logging import getLogger	1✔
13	import os	1✔
14
15	logger = getLogger(__name__)	1✔
16
17
18	class Collector(object):	1✔
19	def __init__(self):	1✔
20	self._init_data()	1✔
21
22	def _init_data(self):	1✔
23	self.accounts = []	1✔
24	self.admins = []	1✔
25	self.terms = {}	1✔
26	self.courses = {}	1✔
27	self.sections = {}	1✔
28	self.enrollments = []	1✔
29	self.enrollment_keys = {}	1✔
30	self.users = {}	1✔
31	self.headers = {	1✔
32	'users': UserHeader(),
33	'admins': AdminHeader(),
34	'terms': TermHeader(),
35	'courses': CourseHeader(),
36	'sections': SectionHeader(),
37	'enrollments': EnrollmentHeader(),
38	}
39
40	def add(self, formatter):	1✔
41	"""
42	Add the passed csv formatter object based on type, returns True if
43	the formatter is added, False otherwise.
44	"""
45	if isinstance(formatter, UserCSV):	1✔
46	return self._add_user(formatter)	×
47	elif isinstance(formatter, EnrollmentCSV):	1✔
48	return self._add_enrollment(formatter)	×
49	elif isinstance(formatter, AdminCSV):	1✔
50	return self._add_admin(formatter)	×
51	elif isinstance(formatter, TermCSV):	1✔
52	return self._add_term(formatter)	×
53	elif isinstance(formatter, CourseCSV):	1✔
54	return self._add_course(formatter)	1✔
55	elif isinstance(formatter, SectionCSV):	1✔
UNCOV 56	return self._add_section(formatter)	×
57	else:
58	raise TypeError(	1✔
59	'Unknown CSVFormat class: {}'.format(type(formatter)))
60
61	def _add_admin(self, formatter):	1✔
62	self.admins.append(formatter)	×
63	return True	×
64
65	def _add_user(self, formatter):	1✔
UNCOV 66	if formatter.key not in self.users:	×
UNCOV 67	self.users[formatter.key] = formatter	×
68	return True	×
UNCOV 69	return False	×
70
71	def _add_term(self, formatter):	1✔
UNCOV 72	if formatter.key not in self.terms:	×
UNCOV 73	self.terms[formatter.key] = formatter	×
74	return True	×
75	return False	×
76
77	def _add_course(self, formatter):	1✔
78	if formatter.key not in self.courses:	1✔
79	self.courses[formatter.key] = formatter	1✔
80	return True	1✔
81	return False	×
82
83	def _add_section(self, formatter):	1✔
84	if formatter.key not in self.sections:	×
UNCOV 85	self.sections[formatter.key] = formatter	×
UNCOV 86	return True	×
UNCOV 87	return False	×
88
89	def _add_enrollment(self, formatter):	1✔
UNCOV 90	if formatter.key not in self.enrollment_keys:	×
UNCOV 91	self.enrollment_keys[formatter.key] = True	×
UNCOV 92	self.enrollments.append(formatter)	×
UNCOV 93	return True	×
UNCOV 94	return False	×
95
96	def has_data(self):	1✔
97	"""
98	Returns True if the collector contains data, False otherwise.
99	"""
100	for csv_type in self.headers:	1✔
101	if len(getattr(self, csv_type)):	1✔
102	return True	1✔
103	return False	1✔
104
105	def write_files(self):	1✔
106	"""
107	Writes all csv files. Returns a path to the csv files, or None
108	if no data was written.
109	"""
110	filepath = None	1✔
111	if self.has_data():	1✔
112	filepath = datetime.now().strftime('%Y/%m/%d/%H%M%S-%f')	1✔
113	for csv_type in self.headers:	1✔
114	try:	1✔
115	data = list(getattr(self, csv_type).values())	1✔
116	data.sort()	1✔
117	except AttributeError:	1✔
118	data = getattr(self, csv_type)	1✔
119
120	if len(data):	1✔
121	os.makedirs(os.path.join(	1✔
122	settings.MEDIA_ROOT, filepath), exist_ok=True)
123	filename = os.path.join(filepath, csv_type + '.csv')	1✔
124	with default_storage.open(filename, mode='w') as f:	1✔
125	headers = self.headers[csv_type]	1✔
126	f.write(str(headers))	1✔
127	for line in data:	1✔
128	f.write(str(line))	1✔
129
130	self._init_data()	1✔
131
132	if getattr(settings, 'TRAINING_IMPORT_CSV_DEBUG', False):	1✔
133	logger.debug('CSV PATH: {}'.format(filepath))	1✔
134	return None	1✔
135	else:
136	return filepath	1✔

uw-it-aca / canvas-training-provisioner / 19684186419

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous