Merge c10b4e9 into 029363f

reana_db/models.py

@@ -13,9 +13,10 @@
 import enum
 import uuid
 
-from sqlalchemy import (Boolean, Column, DateTime, Enum, ForeignKey, Integer,
+from sqlalchemy import (Boolean, Column, DateTime, Enum, Float, ForeignKey,
                         String, Text, UniqueConstraint)
 from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.ext.hybrid import hybrid_property
 from sqlalchemy.orm import relationship
 from sqlalchemy_utils import JSONType, UUIDType
 from sqlalchemy_utils.models import Timestamp
@@ -98,8 +99,10 @@ class Workflow(Base, Timestamp):
     run_started_at = Column(DateTime)
     run_finished_at = Column(DateTime)
     run_stopped_at = Column(DateTime)
-    run_number = Column(Integer)
+    _run_number = Column('run_number', Float)
     job_progress = Column(JSONType, default=dict)
+    workspace_path = Column(String)
+    restart = Column(Boolean, default=False)
     # job_progress = {
     #  jobs_total = {total: job_number}
     #  jobs_running = {job_ids: [], total: c}
@@ -128,7 +131,10 @@ def __init__(self,
                  status=WorkflowStatus.created,
                  git_ref='',
                  git_repo=None,
-                 git_provider=None):
+                 git_provider=None,
+                 workspace_path=None,
+                 restart=False,
+                 run_number=None):
         """Initialize workflow model."""
         self.id_ = id_
         self.name = name
@@ -145,26 +151,50 @@ def __init__(self,
         self.git_ref = git_ref
         self.git_repo = git_repo
         self.git_provider = git_provider
-        from .database import Session
-        last_workflow = Session.query(Workflow).filter_by(
-            name=name,
-            owner_id=owner_id).\
-            order_by(Workflow.run_number.desc()).first()
-        if not last_workflow:
-            self.run_number = 1
-        else:
-            self.run_number = last_workflow.run_number + 1
+        self.restart = restart
+        self._run_number = self.assign_run_number(run_number)
+        self.workspace_path = workspace_path or build_workspace_path(
+            self.owner_id, self.id_)
 
     def __repr__(self):
         """Workflow string represetantion."""
         return '<Workflow %r>' % self.id_
 
-    def get_workspace(self):
-        """Build workflow directory path.
+    @hybrid_property
+    def run_number(self):
+        """Property of run_number."""
+        if self._run_number.is_integer():
+            return int(self._run_number)
+        return self._run_number
 
-        :return: Path to the workflow workspace directory.
-        """
-        return build_workspace_path(self.owner_id, self.id_)
+    @run_number.expression
+    def run_number(cls):
+        from sqlalchemy import func
+        return func.abs(cls._run_number)
+
+    def assign_run_number(self, run_number):
+        """Assing run number."""
+        from .database import Session
+        if run_number:
+            last_workflow = Session.query(Workflow).filter(
+                Workflow.name == self.name,
+                Workflow.run_number >= int(run_number),
+                Workflow.run_number < int(run_number) + 1,
+                Workflow.owner_id == self.owner_id).\
+                order_by(Workflow.run_number.desc()).first()
+        else:
+            last_workflow = Session.query(Workflow).filter_by(
+                name=self.name,
+                restart=False,
+                owner_id=self.owner_id).\
+                order_by(Workflow.run_number.desc()).first()
+        if last_workflow and self.restart:
+            return round(last_workflow.run_number + 0.1, 1)
+        else:
+            if not last_workflow:
+                return 1
+            else:
+                return last_workflow.run_number + 1
 
     def get_input_parameters(self):
         """Return workflow parameters."""
@@ -184,7 +214,7 @@ def get_owner_access_token(self):
 
     def get_full_workflow_name(self):
         """Return full workflow name including run number."""
-        return "{}.{}".format(self.name, self.run_number)
+        return "{}.{}".format(self.name, str(self.run_number))
 
     @staticmethod
     def update_workflow_status(db_session, workflow_uuid, status,

reana_db/utils.py

@@ -88,7 +88,7 @@ def _get_workflow_with_uuid_or_name(uuid_or_name, user_uuid):
 
         # Try to split the dot-separated string.
         try:
-            workflow_name, run_number = uuid_or_name.rsplit('.', maxsplit=1)
+            workflow_name, run_number = uuid_or_name.split('.', maxsplit=1)
         except ValueError:
             # Couldn't split. Probably not a dot-separated string.
             #  -> Search with `uuid_or_name`
@@ -102,7 +102,9 @@ def _get_workflow_with_uuid_or_name(uuid_or_name, user_uuid):
 
         # `run_number` was specified.
         # Check `run_number` is valid.
-        if not run_number.isdigit():
+        try:
+            float(run_number)
+        except ValueError:
             # `uuid_or_name` was split, so it is a dot-separated string
             # but it didn't contain a valid `run_number`.
             # Assume that this dot-separated string is the name of

reana_db/version.py

@@ -14,4 +14,4 @@
 
 from __future__ import absolute_import, print_function
 
-__version__ = "0.7.0.dev20200129"
+__version__ = "0.7.0.dev20200205"

tests/conftest.py

@@ -0,0 +1,55 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of REANA.
+# Copyright (C) 2017, 2018 CERN.
+#
+# REANA is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+
+"""Pytest configuration for REANA-DB."""
+
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import scoped_session, sessionmaker
+from sqlalchemy_utils import create_database, database_exists, drop_database
+
+from reana_db.models import Base
+
+
+@pytest.fixture(scope="module")
+def db():
+    """Flask application fixture."""
+    test_db_engine = create_engine('sqlite:///testdb.db')
+    if not database_exists(test_db_engine.url):
+        create_database(test_db_engine.url)
+    Base.metadata.create_all(bind=test_db_engine)
+    yield test_db_engine
+    drop_database(test_db_engine.url)
+
+
+@pytest.fixture()
+def session(db):
+    """Create a SQL Alchemy session.
+
+    Scope: function
+
+    This fixture offers a SQLAlchemy session which has been created from the
+    ``db_engine`` fixture.
+
+    .. code-block:: python
+
+        from reana_db.models import Workflow
+
+        def test_create_workflow(session):
+            workflow = Workflow(...)
+            session.add(workflow)
+            session.commit()
+    """
+    Session = scoped_session(sessionmaker(autocommit=False,
+                                          autoflush=False,
+                                          bind=db))
+    Base.query = Session.query_property()
+    from reana_db.database import Session as _Session
+    _Session.configure(bind=db)
+    yield Session
+    Session.close()

tests/test_models.py

@@ -0,0 +1,69 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of REANA.
+# Copyright (C) 2019 CERN.
+#
+# REANA is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+
+"""REANA-DB models tests."""
+
+from uuid import uuid4
+
+import pytest
+
+from reana_db.models import Workflow
+
+
+def test_workflow_run_number_assignment(db, session):
+    """Test workflow run number assignment."""
+    workflow_name = 'workflow'
+    owner_id = str(uuid4())
+    first_workflow = Workflow(
+        id_=str(uuid4()),
+        name=workflow_name,
+        owner_id=owner_id,
+        reana_specification=[],
+        type_='serial',
+        logs='',
+    )
+    session.add(first_workflow)
+    session.commit()
+    assert first_workflow.run_number == 1
+    second_workflow = Workflow(
+        id_=str(uuid4()),
+        name=workflow_name,
+        owner_id=owner_id,
+        reana_specification=[],
+        type_='serial',
+        logs='',
+    )
+    session.add(second_workflow)
+    session.commit()
+    assert second_workflow.run_number == 2
+    first_workflow_restart = Workflow(
+        id_=str(uuid4()),
+        name=workflow_name,
+        owner_id=owner_id,
+        reana_specification=[],
+        type_='serial',
+        logs='',
+        restart=True,
+        run_number=first_workflow.run_number,
+    )
+    session.add(first_workflow_restart)
+    session.commit()
+    assert first_workflow_restart.run_number == 1.1
+    first_workflow_second_restart = Workflow(
+        id_=str(uuid4()),
+        name=workflow_name,
+        owner_id=owner_id,
+        reana_specification=[],
+        type_='serial',
+        logs='',
+        restart=True,
+        run_number=first_workflow_restart.run_number,
+    )
+    session.add(first_workflow_second_restart)
+    session.commit()
+    assert first_workflow_second_restart.run_number == 1.2