Improve npm workspace processing

Signed-off-by: Ayan Sinha Mahapatra <[email protected]>

src/packagedcode/npm.py

@@ -69,6 +69,15 @@ def logger_debug(*args):
 
 class BaseNpmHandler(models.DatafileHandler):
 
+    lockfile_names = {
+        'package-lock.json',
+        '.package-lock.json',
+        'npm-shrinkwrap.json',
+        'yarn.lock',
+        'shrinkwrap.yaml',
+        'pnpm-lock.yaml'
+    }
+
     @classmethod
     def assemble(cls, package_data, resource, codebase, package_adder):
         """
@@ -85,19 +94,11 @@ def assemble(cls, package_data, resource, codebase, package_adder):
         If there is no package.json, we do not have a package instance. In this
         case, we yield each of the dependencies in each lock file.
         """
-        lockfile_names = {
-            'package-lock.json',
-            '.package-lock.json',
-            'npm-shrinkwrap.json',
-            'yarn.lock',
-            'shrinkwrap.yaml',
-            'pnpm-lock.yaml'
-        }
 
         package_resource = None
         if resource.name == 'package.json':
             package_resource = resource
-        elif resource.name in lockfile_names:
+        elif resource.name in cls.lockfile_names:
             if resource.has_parent():
                 siblings = resource.siblings(codebase)
                 package_resource = [r for r in siblings if r.name == 'package.json']
@@ -117,10 +118,15 @@ def assemble(cls, package_data, resource, codebase, package_adder):
         pkg_data = package_resource.package_data[0]
         pkg_data = models.PackageData.from_dict(pkg_data)
 
-        workspace_root_path = package_resource.parent(codebase).path
+        workspace_root = package_resource.parent(codebase)
+        workspace_root_path = None
+        if workspace_root:
+            workspace_root_path = package_resource.parent(codebase).path
         workspaces = pkg_data.extra_data.get('workspaces') or []
+
         # Also look for pnpm workspaces
-        if not workspaces:
+        pnpm_workspace = None
+        if not workspaces and workspace_root:
             pnpm_workspace_path = os.path.join(workspace_root_path, 'pnpm-workspace.yaml')
             pnpm_workspace = codebase.get_resource(path=pnpm_workspace_path)
             if pnpm_workspace:
@@ -139,7 +145,7 @@ def assemble(cls, package_data, resource, codebase, package_adder):
         cls.update_workspace_members(workspace_members, codebase)
 
         # do we have enough to create a package?
-        if pkg_data.purl:
+        if pkg_data.purl and not workspaces:
             package = models.Package.from_package_data(
                 package_data=pkg_data,
                 datafile_path=package_resource.path,
@@ -151,35 +157,128 @@ def assemble(cls, package_data, resource, codebase, package_adder):
             # Always yield the package resource in all cases and first!
             yield package
 
-            root = package_resource.parent(codebase)
-            if root:
-                for npm_res in cls.walk_npm(resource=root, codebase=codebase):
+            if workspace_root:
+                for npm_res in cls.walk_npm(resource=workspace_root, codebase=codebase):
                     if package_uid and package_uid not in npm_res.for_packages:
                         package_adder(package_uid, npm_res, codebase)
                     yield npm_res
-            elif codebase.has_single_resource:
-                if package_uid and package_uid not in package_resource.for_packages:
-                    package_adder(package_uid, package_resource, codebase)
             yield package_resource
 
+        elif workspaces:
+            yield from cls.create_packages_from_workspaces(
+                workspace_members=workspace_members,
+                workspace_root=workspace_root,
+                codebase=codebase,
+                package_adder=package_adder,
+                pnpm=pnpm_workspace and pkg_data.purl,
+            )
+
+            package_uid = None
+            if pnpm_workspace and pkg_data.purl:
+                package = models.Package.from_package_data(
+                    package_data=pkg_data,
+                    datafile_path=package_resource.path,
+                )
+                package_uid = package.package_uid
+
+                package.populate_license_fields()
+
+                # Always yield the package resource in all cases and first!
+                yield package
+
+                if workspace_root:
+                    for npm_res in cls.walk_npm(resource=workspace_root, codebase=codebase):
+                        if package_uid and not npm_res.for_packages:
+                            package_adder(package_uid, npm_res, codebase)
+                        yield npm_res
+                yield package_resource
+
         else:
             # we have no package, so deps are not for a specific package uid
             package_uid = None
 
+        yield from cls.yield_npm_dependencies_and_resources(
+            package_resource=package_resource,
+            package_data=pkg_data,
+            package_uid=package_uid,
+            codebase=codebase,
+            package_adder=package_adder,
+        )
+
+    @classmethod
+    def yield_npm_dependencies_and_resources(cls, package_resource, package_data, package_uid, codebase, package_adder):
+
         # in all cases yield possible dependencies
-        yield from yield_dependencies_from_package_data(pkg_data, package_resource.path, package_uid)
+        yield from yield_dependencies_from_package_data(package_data, package_resource.path, package_uid)
 
         # we yield this as we do not want this further processed
         yield package_resource
 
         for lock_file in package_resource.siblings(codebase):
-            if lock_file.name in lockfile_names:
+            if lock_file.name in cls.lockfile_names:
                 yield from yield_dependencies_from_package_resource(lock_file, package_uid)
 
                 if package_uid and package_uid not in lock_file.for_packages:
                     package_adder(package_uid, lock_file, codebase)
                 yield lock_file
 
+    @classmethod
+    def create_packages_from_workspaces(
+        cls,
+        workspace_members,
+        workspace_root,
+        codebase,
+        package_adder,
+        pnpm=False,
+    ):
+
+        workspace_package_uids = []
+        for workspace_member in workspace_members:
+            if not workspace_member.package_data:
+                continue
+
+            pkg_data = workspace_member.package_data[0]
+            pkg_data = models.PackageData.from_dict(pkg_data)
+
+            package = models.Package.from_package_data(
+                package_data=pkg_data,
+                datafile_path=workspace_member.path,
+            )
+            package_uid = package.package_uid
+            workspace_package_uids.append(package_uid)
+
+            package.populate_license_fields()
+
+            # Always yield the package resource in all cases and first!
+            yield package
+
+            member_root = workspace_member.parent(codebase)
+            package_adder(package_uid, member_root, codebase)
+            for npm_res in cls.walk_npm(resource=member_root, codebase=codebase):
+                if package_uid and package_uid not in npm_res.for_packages:
+                    package_adder(package_uid, npm_res, codebase)
+                yield npm_res
+
+            yield from cls.yield_npm_dependencies_and_resources(
+                package_resource=workspace_member,
+                package_data=pkg_data,
+                package_uid=package_uid,
+                codebase=codebase,
+                package_adder=package_adder,
+            )
+
+        # All resources which are not part of a workspace package exclusively
+        # are a part of all packages (this is skipped if we have a root pnpm
+        # package)
+        if pnpm:
+            return
+        for npm_res in cls.walk_npm(resource=workspace_root, codebase=codebase):
+            if npm_res.for_packages:
+                continue
+
+            npm_res.for_packages = workspace_package_uids
+            npm_res.save(codebase)
+
     @classmethod
     def walk_npm(cls, resource, codebase, depth=0):
         """
@@ -1081,6 +1180,7 @@ def parse(cls, location, package_only=False):
                 name, version = name_version.split("@")
             elif major_v == "5" or is_shrinkwrap:
                 if len(sections) == 3:
+                    namespace = None
                     _, name, version = sections
                 elif len(sections) == 4:
                     _, namespace, name, version = sections