Import Irap CSV

hitier
2 parents 02130a91 92adc966
Showing 8 changed files with 433 additions and 93 deletions Show diff stats
CHANGELOG.md
VERSION.txt
app/commands/commands.py
app/models.py
tests/backend_tests.py
tests/cli_tests.py
tests/common_db_feed.py
tests/db_tests.py
@@ -24,6 +24,11 @@ or major refactoring improvments.
  
 ## Unreleased
  
+## [0.3.pre-5] - 2021-04-30 - Import Irap CSV
+### New
+Db models
+Csv reading and feeding
+
 ## [0.3.pre-4] - 2021-04-26 - Export chart 
 ### New
 Export chart as png
-0.3.pre-4
+0.3.pre-5
 import csv
 import os
 import sys
+
 import click
 import random
  
 from flask import current_app
-from sqlalchemy.exc import OperationalError, IntegrityError
-from sqlalchemy.orm.exc import NoResultFound
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.sql import func
-from sqlalchemy.ext.automap import automap_base
-from sqlalchemy.orm import Session
-from sqlalchemy import create_engine
  
-from app.models import db, Agent, Service, Project, Capacity, Period, Charge
+from app.models import db, Agent, Service, Project, Capacity, Period, Charge, AgentStatus, Company, AgentBap, \
+    AgentGrade, Category, Label, ProjectLabel
+# TODO: rename to methods and add get_roles()
 from app.auth.models import User, _nameToRole, _roleToName
  
 from . import bp
@@ -57,38 +56,129 @@ def feed_from_irap(csv_file_name):
     with open(csv_file_name, newline='') as csvfile:
         csvreader = csv.DictReader(csvfile, delimiter=',', quotechar='"')
         for row in csvreader:
-            # print('\n'.join(row.keys()))
-            # break
             # Remove any leading/trailing spaces
             row = {k: v.strip() for k, v in row.items()}
             rows.append(row)
  
     firstname_key = 'NOM'
-    secondname_key = 'prénom'
+    secondname_key = 'PRÉNOM'
+    status_key = 'STATUT'
+    virtual_key = 'VIRTUEL'
+    company_key = 'SOCIÉTÉ'
+    bap_key = 'BAP'
+    grade_key = 'GRADE CORPS'
     project_key = 'PROJETS'
-    service_key = 'Groupe métier'
-    # typology_title = 'TYPOLOGIE'
-    # thematic_title = 'thématique'
+    service_key = 'GROUPE MÉTIER'
+    categorie_keys = ['TYPOLOGIE', 'THÉMATIQUE']
  
     # Get the columns values
     #
-    projects = [r[project_key] for r in rows]
-    projects = sorted(set(projects))
-    agents = [(r[firstname_key], r[secondname_key].strip()) for r in rows]
-    agents = sorted(set(agents))
-    services = [r[service_key] for r in rows]
-    services = sorted(set(services))
-
-    # thematics = [r[thematic_title] for r in rows]
-    # thematics = sorted(set(thematics))
-    # typologies = [r[typology_title] for r in rows]
-    # typologies = sorted(set(typologies))
-
-    # Feed agents from column
+    agents = []
+    services = []
+    baps = []
+    grades = []
+    companies = []
+    statuses = []
+
+    # build a category dict of lists
+    #   key being the category name,
+    #   list being filled with corresponding labels
+    categories = {k: [] for k in categorie_keys}
+
     #
-    for a in agents:
-        n_a = Agent(firstname=a[0], secondname=a[1])
-        db.session.add(n_a)
+    categorized_projects = {}
+
+    for r in rows:
+        services.append(r[service_key])
+        baps.append(r[bap_key])
+        grades.append(r[grade_key])
+        companies.append(r[company_key])
+        statuses.append(r[status_key])
+
+        # categorized_projects
+        project = r[project_key]
+        categorized_projects[project] = []
+
+        # fill in the category-labels dict
+        for k in categorie_keys:
+            categories[k].append(r[k])
+            categorized_projects[project].append(r[k])
+
+        # build agents list of dicts
+        agents.append({
+            'firstname': r[firstname_key],
+            'secondname': r[secondname_key],
+            'status': r[status_key],
+            'virtual': r[virtual_key],
+            'company': r[company_key],
+            'bap': r[bap_key],
+            'grade': r[grade_key]})
+
+    # Uppercase the small tables
+    #
+    baps = [x.upper() for x in baps]
+    grades = [x.upper() for x in grades]
+    statuses = [x.upper() for x in statuses]
+
+    # Now, sort the lists
+    #
+    #  1- first remove empty string with filter()
+    #  2- then keep only uniq item with set()
+    #  3- at last alpha sort  with sorted()
+    #
+    services = sorted(set(filter(None, services)))
+    baps = sorted(set(filter(None, baps)))
+    grades = sorted(set(filter(None, grades)))
+    companies = sorted(set(filter(None, companies)))
+    statuses = sorted(set(filter(None, statuses)))
+
+    # Do the same for the projects, that are keys of categorized_projects
+    #
+    projects = sorted(set(categorized_projects.keys()))
+
+    # Do the same for the category labels stored as a dict
+    #
+    #   k is category name
+    #   v is labels list for that category
+    for k in categorie_keys:
+        labels = sorted(set(filter(None, categories[k])))
+        categories[k] = labels
+
+    # At least, as agents is a list of dicts, sorting is a bit tricky
+    #
+    # the first one liner will store the last agent's line only
+    # then we alpha sort  on the name
+    # on both, the discrimination is based on the name couple: (firstname, secondname)
+    #
+    agents = list({(a['firstname'], a['secondname']): a for a in agents}.values())
+    agents = sorted(agents, key=lambda a: (a['firstname'], a['secondname']))
+
+    # Feed baps from column
+    #
+    for b in baps:
+        n_b = AgentBap(name=b)
+        db.session.add(n_b)
+    db.session.commit()
+
+    # Feed grades from column
+    #
+    for g in grades:
+        n_g = AgentGrade(name=g)
+        db.session.add(n_g)
+    db.session.commit()
+
+    # Feed companies from column
+    #
+    for c in companies:
+        n_c = Company(name=c)
+        db.session.add(n_c)
+    db.session.commit()
+
+    # Feed statuses from column
+    #
+    for s in statuses:
+        n_s = AgentStatus(name=s)
+        db.session.add(n_s)
     db.session.commit()
  
     # Feed projects from column
@@ -107,7 +197,8 @@ def feed_from_irap(csv_file_name):
  
     # Feed periods names
     # Todo: are statically built,
-    #       should come from year column name.
+    #      should come from year column name.
+    #      also see later
     #
     for p in range(2011, 2030):
         n_p = Period(name=f"{p}")
@@ -115,9 +206,65 @@ def feed_from_irap(csv_file_name):
     db.session.commit()
  
     # Add one default capacity
-    db.session.add(Capacity(name="Travailleur"))
+    #
+    db.session.add(Capacity(name="Agent"))
+    db.session.commit()
+
+    # Feed categories and labels
+    #
+    for c, l in categories.items():
+        current_app.logger.debug("Adding category " + c)
+        n_c = Category(name=c)
+        db.session.add(n_c)
+        # db.session.commit()
+        for label in l:
+            current_app.logger.debug("Adding label {} for id {}".format(label, n_c.id))
+            n_l = Label(name=label, category=n_c)
+            db.session.add(n_l)
+    db.session.commit()
+
+    # Feed project's labels
+    #
+    for p, labels in categorized_projects.items():
+        n_p = Project.query.filter(Project.name == p).one()
+        for l in labels:
+            n_l = Label.query.filter(Label.name == l).one()
+            n_c = n_l.category
+            n_pl = ProjectLabel(project=n_p, category=n_c, label=n_l)
+            db.session.add(n_pl)
     db.session.commit()
  
+    # Feed agents from agents list previously filled
+    #
+    for a in agents:
+        status = AgentStatus.query.filter(AgentStatus.name == a['status']).one_or_none()
+        if status is None and a['status']:
+            status = AgentStatus(name=a['status'])
+        company = Company.query.filter(Company.name == a['company']).one_or_none()
+        if company is None and a['company']:
+            company = Company(name=a['company'])
+        bap = AgentBap.query.filter(AgentBap.name == a['bap']).one_or_none()
+        if bap is None and a['bap']:
+            bap = AgentBap(name=a['bap'])
+        grade = AgentGrade.query.filter(AgentGrade.name == a['grade']).one_or_none()
+        if grade is None and a['grade']:
+            grade = AgentBap(name=a['grade'])
+        virtual = 1 if a['virtual'] else 0
+        permanent = 1 if a['status'] == 'PERM' else 0
+        n_a = Agent(firstname=a['firstname'],
+                    secondname=a['secondname'],
+                    status_id=status.id if status else None,
+                    company_id=company.id if company else None,
+                    bap_id=bap.id if bap else None,
+                    grade_id=grade.id if grade else None,
+                    virtual=virtual,
+                    permanent=permanent)
+        db.session.add(n_a)
+    db.session.commit()
+
+    # Feed agents from agents list previously filled
+    #
+
     # Now feed the charges.
     #
     # At least one for each csv row
@@ -128,6 +275,7 @@ def feed_from_irap(csv_file_name):
         a = Agent.query.filter(Agent.firstname == r[firstname_key], Agent.secondname == r[secondname_key]).one()
         s = Service.query.filter(Service.name == r[service_key]).one()
         c = Capacity.query.first()
+        # TODO: period names should come from db request
         for period_name in range(2011, 2030):
             t = Period.query.filter(Period.name == period_name).one()
             charge = r[f"{period_name}"]
@@ -323,8 +471,8 @@ def feed_random_charges(agent):
 def user_add(email, name, login, password, role):
     """ Add a new user in db."""
     user = User.query.filter(User.name == name).one_or_none()
-    if (user):
-        current_app.logger.error(f"user already exists {name}")
+    if user:
+        current_app.logger.warn(f"user already exists {name}")
         return
     user = User(email=email, name=name, login=login, password=password, role=role)
     db.session.add(user)
@@ -351,12 +499,12 @@ def user_update(user_id, name, role, email, password):
         user.set_role(role)
         print(f"User --{user.name}-- role updated to {_roleToName[user.role]}")
     if email:
-        user.email=email
+        user.email = email
         print(f"User --{user.name}-- email updated to {user.email}")
     if password:
         print(f"User --{user.name}-- password updated")
         user.set_password(password)
-    if not ( name or role or email or password):
+    if not (name or role or email or password):
         print(f"No update for user --{user.name}--")
     db.session.commit()
  
 from flask_sqlalchemy import SQLAlchemy
+from sqlalchemy.orm import relationship
  
 db = SQLAlchemy()
  
  
-class Agent(db.Model):
+#
+# Categorized projects
+#
+
+class Project(db.Model):
     id = db.Column(db.Integer, primary_key=True)
-    firstname = db.Column(db.String(100))
-    secondname = db.Column(db.String(100))
+    name = db.Column(db.String)
+    labels = relationship("ProjectLabel", back_populates="project")
  
  
-class Project(db.Model):
+class Category(db.Model):
     id = db.Column(db.Integer, primary_key=True)
-    name = db.Column(db.String(100), unique=True)
+    name = db.Column(db.String)
+    labels = relationship("Label", back_populates="category")
+    projects = relationship("ProjectLabel", back_populates="category")
  
  
-class Service(db.Model):
+class Label(db.Model):
     id = db.Column(db.Integer, primary_key=True)
-    name = db.Column(db.String(100), unique=True)
-    abbr = db.Column(db.String(50), unique=True)
+    name = db.Column(db.String, unique=True)
+    category_id = db.Column(db.Integer, db.ForeignKey('category.id'))
+    category = relationship("Category", back_populates="labels")
+    projects = relationship("ProjectLabel", back_populates="label")
+
+
+class ProjectLabel(db.Model):
+    project_id = db.Column(db.Integer, db.ForeignKey('project.id'), primary_key=True)
+    category_id = db.Column(db.Integer, db.ForeignKey('category.id'), primary_key=True)
+    label_id = db.Column(db.Integer, db.ForeignKey('label.id'))
+
+    project = relationship("Project", back_populates="labels")
+    category = relationship("Category", back_populates="projects")
+    label = relationship("Label", back_populates="projects")
+
+
+#
+# Agents
+#
+
+class AgentBap(db.Model):
+    id = db.Column(db.Integer, primary_key=True)
+    name = db.Column(db.String(16))
  
  
-class Function(db.Model):
+class AgentGrade(db.Model):
+    id = db.Column(db.Integer, primary_key=True)
+    name = db.Column(db.String(16))
+
+
+class AgentStatus(db.Model):
+    id = db.Column(db.Integer, primary_key=True)
+    name = db.Column(db.String(16))
+
+
+class Company(db.Model):
+    id = db.Column(db.Integer, primary_key=True)
+    name = db.Column(db.String(16))
+
+
+class Agent(db.Model):
+    id = db.Column(db.Integer, primary_key=True)
+    firstname = db.Column(db.String(100))
+    secondname = db.Column(db.String(100))
+    company_id = db.Column(db.Integer, db.ForeignKey('company.id'))
+    grade_id = db.Column(db.Integer, db.ForeignKey('agent_grade.id'))
+    status_id = db.Column(db.Integer, db.ForeignKey('agent_status.id'))
+    bap_id = db.Column(db.Integer, db.ForeignKey('agent_bap.id'))
+    virtual = db.Column(db.Integer)  # integer boolean
+    permanent = db.Column(db.Integer)  # integer boolean
+
+
+class Service(db.Model):
     id = db.Column(db.Integer, primary_key=True)
     name = db.Column(db.String(100), unique=True)
+    abbr = db.Column(db.String(50), unique=True)
  
  
 class Capacity(db.Model):
+import os
+import sys
 import unittest
+from shutil import copyfile
  
 from pdc_config import TestConfig
 from app import create_app, db_mgr, db
@@ -7,18 +10,29 @@ from app.auth.models import User
  
 class BaseTestCase(unittest.TestCase):
     def setUp(self):
-        # configure data base
+        # initialise app
         self.app = create_app(TestConfig)
+
+        # copy resource  demo db to test file
+        appdir = os.path.join(os.path.dirname(os.path.abspath(__file__)), os.pardir)
+        sqlite_file_name = os.path.abspath(os.path.join(appdir, 'resources', 'lesia-btp.sqlite'))
+        if not os.path.isdir(self.app.instance_path):
+            os.mkdir(self.app.instance_path)
+        self.db_path = os.path.join(self.app.instance_path, 'test.db')
+        copyfile(sqlite_file_name, self.db_path)
+
+        # force db path to newly create file
+        self.app.config.update(
+            SQLALCHEMY_DATABASE_URI='sqlite:///' + self.db_path
+        )
+
+        # update flask context
         self.app_context = self.app.app_context()
         self.app_context.push()
-        # TODO: shall we always copy db from resources sqlite file ?
-        #       that would allow us to run all tests in sqlite:memory:
-        # db.create_all()
-        # admin = User(email='admin@nowhere.org', name='admin', login='admin', password='admin', role='admin')
-        # db.session.add(admin)
-        # db.session.commit()
  
     def tearDown(self):
+        if os.path.isfile(self.db_path):
+            os.remove(self.db_path)
         self.app_context.pop()
  
     def test_always_true(self):
@@ -46,45 +60,3 @@ class DbMgrTestCase(BaseTestCase):
         # Waiting for 17 periods + headers line
         self.assertEqual(18, len(stacked_charges))
  
-
-class AuthModelTestCase(BaseTestCase):
-
-    def skip_if_no_sqlitememory(self):
-        if 'memory' not in self.app.config['SQLALCHEMY_DATABASE_URI']:
-            self.skipTest("Needs in memory sqlite")
-
-    def get_admin(self):
-        return User.query.filter(User.name == 'admin').one()
-
-    def setUp(self):
-        BaseTestCase.setUp(self)
-        self.skip_if_no_sqlitememory()
-        db.create_all()
-        admin = User(email='admin@nowhere.org', name='admin', login='admin', role='admin')
-        db.session.add(admin)
-        db.session.commit()
-
-    def test_in_memory(self):
-        self.app.logger.info("In memory Sqlite DB for tests")
-        self.assertTrue(True)
-
-    def test_setrole(self):
-        admin = self.get_admin()
-        admin.set_role("ADMIN")
-        db.session.commit()
-        admin = self.get_admin()
-        self.assertTrue(admin is not None)
-        self.assertTrue(admin.has_role("ADMIN"))
-        self.assertFalse(admin.has_role("SERVICE"))
-
-    def test_setrole_valueerror(self):
-        admin = self.get_admin()
-        with self.assertRaises(ValueError) as ve:
-            admin.set_role("NOSUCHROLE")
-
-    def test_setcheckpassword(self):
-        admin = self.get_admin()
-        admin.set_password("hahaha")
-        db.session.commit()
-        admin2 = self.get_admin()
-        self.assertTrue(admin2.check_password("hahaha"))
@@ -0,0 +1,49 @@
+import unittest
+
+from app import create_app, db, User
+from app.commands.commands import feed_from_irap, show_roles, user_add, user_show_all
+from pdc_config import TestConfig
+
+from app.commands import bp as cli_bp
+
+
+class CliBaseTestCase(unittest.TestCase):
+
+    def setUp(self):
+        # Get rid of a strange 'ValueError: I/O operation' when app.logger outputs
+        TestConfig.PDC_LOGS_LEVEL = 'ERROR'
+        # Force sqlite in memory db
+        TestConfig.SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
+        # Create and register app
+        self.app = create_app(TestConfig)
+        self.app.register_blueprint(cli_bp)
+        self.app_context = self.app.app_context()
+        self.app_context.push()
+        db.create_all()
+
+    def tearDown(self):
+        pass
+
+    def test_hello3(self):
+        runner = self.app.test_cli_runner()
+        result = runner.invoke(user_show_all)
+        print("\n")
+        print(result.output)
+
+    def test_hello2(self):
+        runner = self.app.test_cli_runner()
+        result = runner.invoke(show_roles)
+        print(result.output)
+
+    def test_add_user(self):
+        runner = self.app.test_cli_runner()
+
+        all_users = User.query.all()
+        self.assertEqual(0, len(all_users))
+        # invoke the command directly
+        arguments = ['geo@ici.fr', 'Joseph Hitier', 'joh', 'passwd', 'PUBLIC']
+        result = runner.invoke(user_add, arguments)
+        print(result.output)
+        all_users = User.query.all()
+        print(len(all_users))
+        # self.assertEqual(1, len(all_users))
@@ -0,0 +1,39 @@
+from app.models import Category, db, Label, Project, ProjectLabel
+
+categorized_labels = {'pole': ['Spatial', 'Sol'], 'domaine': ['soleil-terre', 'atmosphere', 'r&t', 'géologie']}
+projects_categories = {'ChemCam': {'pole': 'Spatial', 'domaine': 'géologie'},
+                       'Pilot': {'pole': 'Spatial', 'domaine': 'atmosphere'},
+                       'Ambre': {'pole': 'Spatial', 'domaine': 'soleil-terre'}}
+
+
+def feed_projects():
+    for c_name, labels in categorized_labels.items():
+        n_c = Category(name=c_name)
+        db.session.add(n_c)
+        for l_name in labels:
+            n_l = Label(name=l_name, category=n_c)
+            db.session.add(n_l)
+
+    db.session.commit()
+
+    project_names = projects_categories.keys()
+
+    for p_name in set(project_names):
+        n_p = Project(name=p_name)
+        db.session.add(n_p)
+    db.session.commit()
+
+    for p, categories in projects_categories.items():
+        n_p = db.session.query(Project).filter(Project.name == p).one()
+        for c_name, l_name in categories.items():
+            n_c = db.session.query(Category).filter(Category.name == c_name).one()
+            n_l = db.session.query(Label).filter(Label.name == l_name).one()
+            n_pc = ProjectLabel(project=n_p, category=n_c, label=n_l)
+            db.session.add(n_pc)
+            
+    db.session.commit()
+
+
+def show_categories():
+    for c in db.session.query(Category).all():
+        print(c.name + ": ", ",".join([l.name for l in c.categorized_labels]))
@@ -0,0 +1,71 @@
+import unittest
+
+from app import create_app, db, User
+from app.models import Project
+from pdc_config import TestConfig
+from tests.common_db_feed import feed_projects
+
+
+class DbBaseTestCase(unittest.TestCase):
+    def setUp(self):
+        self.app = create_app(TestConfig)
+        # force db uri to sqlite memory
+        self.app.config.update(
+            SQLALCHEMY_DATABASE_URI='sqlite:///:memory:'
+        )
+        self.app_context = self.app.app_context()
+        self.app_context.push()
+        db.create_all()
+        feed_projects()
+
+    def tearDown(self):
+        db.session.remove()
+        db.drop_all()
+        self.app_context.pop()
+
+    def test_first(self):
+        projects = Project.query.all()
+        self.assertEqual(3, len(projects))
+
+
+class AuthModelTestCase(DbBaseTestCase):
+
+    def skip_if_no_sqlitememory(self):
+        if 'memory' not in self.app.config['SQLALCHEMY_DATABASE_URI']:
+            self.skipTest("Needs in memory sqlite")
+
+    def get_admin(self):
+        return User.query.filter(User.name == 'admin').one()
+
+    def setUp(self):
+        DbBaseTestCase.setUp(self)
+        self.skip_if_no_sqlitememory()
+        db.create_all()
+        admin = User(email='admin@nowhere.org', name='admin', login='admin', role='admin')
+        db.session.add(admin)
+        db.session.commit()
+
+    def test_in_memory(self):
+        self.app.logger.info("In memory Sqlite DB for tests")
+        self.assertTrue(True)
+
+    def test_setrole(self):
+        admin = self.get_admin()
+        admin.set_role("ADMIN")
+        db.session.commit()
+        admin = self.get_admin()
+        self.assertTrue(admin is not None)
+        self.assertTrue(admin.has_role("ADMIN"))
+        self.assertFalse(admin.has_role("SERVICE"))
+
+    def test_setrole_valueerror(self):
+        admin = self.get_admin()
+        with self.assertRaises(ValueError) as ve:
+            admin.set_role("NOSUCHROLE")
+
+    def test_setcheckpassword(self):
+        admin = self.get_admin()
+        admin.set_password("hahaha")
+        db.session.commit()
+        admin2 = self.get_admin()
+        self.assertTrue(admin2.check_password("hahaha"))